Began work on overhauling the deck_builder

This commit is contained in:
mwisnowski 2025-01-14 12:07:49 -08:00
parent e0dd09adee
commit 319f7848d3
10 changed files with 2589 additions and 761 deletions

331
builder_utils.py Normal file
View file

@ -0,0 +1,331 @@
from typing import Dict, List, Tuple, Optional, Any, Callable, TypeVar, Union
import logging
import functools
import time
import pandas as pd
from fuzzywuzzy import process
from settings import (
COMMANDER_CSV_PATH,
FUZZY_MATCH_THRESHOLD,
MAX_FUZZY_CHOICES,
COMMANDER_CONVERTERS,
DATAFRAME_VALIDATION_RULES,
DATAFRAME_VALIDATION_TIMEOUT,
DATAFRAME_BATCH_SIZE,
DATAFRAME_TRANSFORM_TIMEOUT,
DATAFRAME_REQUIRED_COLUMNS
)
from exceptions import (
DeckBuilderError,
CSVValidationError,
DataFrameValidationError,
DataFrameTimeoutError,
EmptyDataFrameError
)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Type variables for generic functions
T = TypeVar('T')
DataFrame = TypeVar('DataFrame', bound=pd.DataFrame)
def timeout_wrapper(timeout: float) -> Callable:
"""Decorator to add timeout to functions.
Args:
timeout: Maximum execution time in seconds
Returns:
Decorated function with timeout
Raises:
DataFrameTimeoutError: If operation exceeds timeout
"""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> T:
start_time = time.time()
result = func(*args, **kwargs)
elapsed = time.time() - start_time
if elapsed > timeout:
raise DataFrameTimeoutError(
func.__name__,
timeout,
elapsed,
{'args': args, 'kwargs': kwargs}
)
return result
return wrapper
return decorator
def get_validation_rules(data_type: str) -> Dict[str, Dict[str, Any]]:
"""Get validation rules for specific data type.
Args:
data_type: Type of data to get rules for
Returns:
Dictionary of validation rules
"""
from settings import (
CREATURE_VALIDATION_RULES,
SPELL_VALIDATION_RULES,
LAND_VALIDATION_RULES
)
rules_map = {
'creature': CREATURE_VALIDATION_RULES,
'spell': SPELL_VALIDATION_RULES,
'land': LAND_VALIDATION_RULES
}
return rules_map.get(data_type, DATAFRAME_VALIDATION_RULES)
@timeout_wrapper(DATAFRAME_VALIDATION_TIMEOUT)
def validate_dataframe(df: pd.DataFrame, rules: Dict[str, Dict[str, Any]]) -> bool:
"""Validate DataFrame against provided rules.
Args:
df: DataFrame to validate
rules: Validation rules to apply
Returns:
True if validation passes
Raises:
DataFrameValidationError: If validation fails
"""
#print(df.columns)
if df.empty:
raise EmptyDataFrameError("validate_dataframe")
try:
validate_required_columns(df)
validate_column_types(df, rules)
return True
except Exception as e:
raise DataFrameValidationError(
"DataFrame validation failed",
{'rules': rules, 'error': str(e)}
)
def validate_column_types(df: pd.DataFrame, rules: Dict[str, Dict[str, Any]]) -> bool:
"""Validate column types against rules.
Args:
df: DataFrame to validate
rules: Type validation rules
Returns:
True if validation passes
Raises:
DataFrameValidationError: If type validation fails
"""
for col, rule in rules.items():
if col not in df.columns:
continue
expected_type = rule.get('type')
if not expected_type:
continue
if isinstance(expected_type, tuple):
valid = any(df[col].dtype.name.startswith(t) for t in expected_type)
else:
valid = df[col].dtype.name.startswith(expected_type)
if not valid:
raise DataFrameValidationError(
col,
rule,
{'actual_type': df[col].dtype.name}
)
return True
def validate_required_columns(df: pd.DataFrame) -> bool:
"""Validate presence of required columns.
Args:
df: DataFrame to validate
Returns:
True if validation passes
Raises:
DataFrameValidationError: If required columns are missing
"""
#print(df.columns)
missing = set(DATAFRAME_REQUIRED_COLUMNS) - set(df.columns)
if missing:
raise DataFrameValidationError(
"missing_columns",
{'required': DATAFRAME_REQUIRED_COLUMNS},
{'missing': list(missing)}
)
return True
@timeout_wrapper(DATAFRAME_TRANSFORM_TIMEOUT)
def process_dataframe_batch(df: pd.DataFrame, batch_size: int = DATAFRAME_BATCH_SIZE) -> pd.DataFrame:
"""Process DataFrame in batches.
Args:
df: DataFrame to process
batch_size: Size of each batch
Returns:
Processed DataFrame
Raises:
DataFrameTimeoutError: If processing exceeds timeout
"""
processed_dfs = []
for i in range(0, len(df), batch_size):
batch = df.iloc[i:i + batch_size].copy()
processed = transform_dataframe(batch)
processed_dfs.append(processed)
return pd.concat(processed_dfs, ignore_index=True)
def transform_dataframe(df: pd.DataFrame) -> pd.DataFrame:
"""Apply transformations to DataFrame.
Args:
df: DataFrame to transform
Returns:
Transformed DataFrame
"""
df = df.copy()
# Fill missing values
df['colorIdentity'] = df['colorIdentity'].fillna('COLORLESS')
df['colors'] = df['colors'].fillna('COLORLESS')
# Convert types
numeric_cols = ['manaValue', 'edhrecRank']
for col in numeric_cols:
if col in df.columns:
df[col] = pd.to_numeric(df[col], errors='coerce')
return df
def combine_dataframes(dfs: List[pd.DataFrame]) -> pd.DataFrame:
"""Combine multiple DataFrames with validation.
Args:
dfs: List of DataFrames to combine
Returns:
Combined DataFrame
Raises:
EmptyDataFrameError: If no valid DataFrames to combine
"""
if not dfs:
raise EmptyDataFrameError("No DataFrames to combine")
valid_dfs = []
for df in dfs:
try:
if validate_dataframe(df, DATAFRAME_VALIDATION_RULES):
valid_dfs.append(df)
except DataFrameValidationError as e:
logger.warning(f"Skipping invalid DataFrame: {e}")
if not valid_dfs:
raise EmptyDataFrameError("No valid DataFrames to combine")
return pd.concat(valid_dfs, ignore_index=True)
def load_commander_data(csv_path: str = COMMANDER_CSV_PATH,
converters: Dict = COMMANDER_CONVERTERS) -> pd.DataFrame:
"""Load and prepare commander data from CSV file.
Args:
csv_path (str): Path to commander CSV file. Defaults to COMMANDER_CSV_PATH.
converters (Dict): Column converters for CSV loading. Defaults to COMMANDER_CONVERTERS.
Returns:
pd.DataFrame: Processed commander dataframe
Raises:
DeckBuilderError: If CSV file cannot be loaded or processed
"""
try:
df = pd.read_csv(csv_path, converters=converters)
df['colorIdentity'] = df['colorIdentity'].fillna('COLORLESS')
df['colors'] = df['colors'].fillna('COLORLESS')
return df
except FileNotFoundError:
logger.error(f"Commander CSV file not found at {csv_path}")
raise DeckBuilderError(f"Commander data file not found: {csv_path}")
except Exception as e:
logger.error(f"Error loading commander data: {e}")
raise DeckBuilderError(f"Failed to load commander data: {str(e)}")
def process_fuzzy_matches(card_name: str,
df: pd.DataFrame,
threshold: int = FUZZY_MATCH_THRESHOLD,
max_choices: int = MAX_FUZZY_CHOICES) -> Tuple[str, List[Tuple[str, int]], bool]:
"""Process fuzzy matching for commander name selection.
Args:
card_name (str): Input card name to match
df (pd.DataFrame): Commander dataframe to search
threshold (int): Minimum score for direct match. Defaults to FUZZY_MATCH_THRESHOLD.
max_choices (int): Maximum number of choices to return. Defaults to MAX_FUZZY_CHOICES.
Returns:
Tuple[str, List[Tuple[str, int]], bool]: Selected card name, list of matches with scores, and match status
"""
try:
match, score, _ = process.extractOne(card_name, df['name'])
if score >= threshold:
return match, [], True
fuzzy_choices = process.extract(card_name, df['name'], limit=max_choices)
fuzzy_choices = [(name, score) for name, score in fuzzy_choices]
return "", fuzzy_choices, False
except Exception as e:
logger.error(f"Error in fuzzy matching: {e}")
raise DeckBuilderError(f"Failed to process fuzzy matches: {str(e)}")
def validate_commander_selection(df: pd.DataFrame, commander_name: str) -> Dict:
"""Validate and format commander data from selection.
Args:
df (pd.DataFrame): Commander dataframe
commander_name (str): Selected commander name
Returns:
Dict: Formatted commander data dictionary
Raises:
DeckBuilderError: If commander data is invalid or missing
"""
try:
filtered_df = df[df['name'] == commander_name]
if filtered_df.empty:
raise DeckBuilderError(f"No commander found with name: {commander_name}")
commander_dict = filtered_df.to_dict('list')
# Validate required fields
required_fields = ['name', 'type', 'colorIdentity', 'colors', 'manaCost', 'manaValue']
for field in required_fields:
if field not in commander_dict or not commander_dict[field]:
raise DeckBuilderError(f"Missing required commander data: {field}")
return commander_dict
except Exception as e:
logger.error(f"Error validating commander selection: {e}")
raise DeckBuilderError(f"Failed to validate commander selection: {str(e)}")

File diff suppressed because it is too large Load diff

View file

@ -1,64 +1,237 @@
"""Custom exceptions for MTG Python Deckbuilder setup operations."""
"""Custom exceptions for the MTG Python Deckbuilder application."""
class MTGSetupError(Exception):
"""Base exception class for MTG setup-related errors."""
pass
class DeckBuilderError(Exception):
"""Base exception class for deck builder errors.
Attributes:
code (str): Error code for identifying the error type
message (str): Descriptive error message
details (dict): Additional error context and details
"""
def __init__(self, message: str, code: str = "DECK_ERR", details: dict | None = None):
"""Initialize the base deck builder error.
Args:
message: Human-readable error description
code: Error code for identification and handling
details: Additional context about the error
"""
self.code = code
self.message = message
self.details = details or {}
super().__init__(self.message)
def __str__(self) -> str:
"""Format the error message with code and details."""
error_msg = f"[{self.code}] {self.message}"
if self.details:
error_msg += f"\nDetails: {self.details}"
return error_msg
class MTGSetupError(DeckBuilderError):
"""Base exception class for MTG setup-related errors.
This exception serves as the base for all setup-related errors in the deck builder,
including file operations, data processing, and validation during setup.
"""
def __init__(self, message: str, code: str = "SETUP_ERR", details: dict | None = None):
"""Initialize the base setup error.
Args:
message: Human-readable error description
code: Error code for identification and handling
details: Additional context about the error
"""
super().__init__(message, code=code, details=details)
class CSVFileNotFoundError(MTGSetupError):
"""Exception raised when a required CSV file is not found.
This exception is raised when attempting to access or process a CSV file
that does not exist in the expected location.
Args:
message: Explanation of the error
filename: Name of the missing CSV file
"""
def __init__(self, message: str, filename: str) -> None:
self.filename = filename
super().__init__(f"{message}: {filename}")
def __init__(self, filename: str, details: dict | None = None):
"""Initialize CSV file not found error.
Args:
filename: Name of the missing CSV file
details: Additional context about the missing file
"""
message = f"Required CSV file not found: '{filename}'"
super().__init__(message, code="CSV_MISSING", details=details)
class MTGJSONDownloadError(MTGSetupError):
"""Exception raised when downloading data from MTGJSON fails.
This exception is raised when there are issues downloading card data
from the MTGJSON API, such as network errors or API failures.
Args:
message: Explanation of the error
url: The URL that failed to download
status_code: HTTP status code if available
"""
def __init__(self, message: str, url: str, status_code: int = None) -> None:
self.url = url
self.status_code = status_code
def __init__(self, url: str, status_code: int | None = None, details: dict | None = None):
"""Initialize MTGJSON download error.
Args:
url: The URL that failed to download
status_code: HTTP status code if available
details: Additional context about the download failure
"""
status_info = f" (Status: {status_code})" if status_code else ""
super().__init__(f"{message}: {url}{status_info}")
message = f"Failed to download from MTGJSON: {url}{status_info}"
super().__init__(message, code="MTGJSON_ERR", details=details)
class DataFrameProcessingError(MTGSetupError):
"""Exception raised when DataFrame operations fail during setup.
# Input Handler Exceptions
class EmptyInputError(DeckBuilderError):
"""Raised when text input validation fails due to empty or whitespace-only input.
This exception is raised when there are issues processing card data
in pandas DataFrames, such as filtering, sorting, or transformation errors.
Args:
message: Explanation of the error
operation: The DataFrame operation that failed (e.g., 'color_filtering', 'commander_processing')
details: Additional error details
Examples:
>>> raise DataFrameProcessingError(
... "Invalid color identity",
... "color_filtering",
... "Color 'P' is not a valid MTG color"
... )
This exception is used by the validate_text method when checking user input.
"""
def __init__(self, message: str, operation: str, details: str = None) -> None:
self.operation = operation
self.details = details
error_info = f" - {details}" if details else ""
super().__init__(f"{message} during {operation}{error_info}")
def __init__(self, field_name: str = "input", details: dict | None = None):
"""Initialize empty input error.
Args:
field_name: Name of the input field that was empty
details: Additional context about the validation failure
"""
message = f"Empty or whitespace-only {field_name} is not allowed"
super().__init__(message, code="EMPTY_INPUT", details=details)
class InvalidNumberError(DeckBuilderError):
"""Raised when number input validation fails.
This exception is used by the validate_number method when checking numeric input.
"""
def __init__(self, value: str, details: dict | None = None):
"""Initialize invalid number error.
Args:
value: The invalid input value
details: Additional context about the validation failure
"""
message = f"Invalid number format: '{value}'"
super().__init__(message, code="INVALID_NUM", details=details)
class InvalidQuestionTypeError(DeckBuilderError):
"""Raised when an unsupported question type is used in the questionnaire method.
This exception is raised when the questionnaire method receives an unknown question type.
"""
def __init__(self, question_type: str, details: dict | None = None):
"""Initialize invalid question type error.
Args:
question_type: The unsupported question type
details: Additional context about the error
"""
message = f"Unsupported question type: '{question_type}'"
super().__init__(message, code="INVALID_QTYPE", details=details)
class MaxAttemptsError(DeckBuilderError):
"""Raised when maximum input attempts are exceeded.
This exception is used when user input validation fails multiple times.
"""
def __init__(self, max_attempts: int, input_type: str = "input", details: dict | None = None):
"""Initialize maximum attempts error.
Args:
max_attempts: Maximum number of attempts allowed
input_type: Type of input that failed validation
details: Additional context about the attempts
"""
message = f"Maximum {input_type} attempts ({max_attempts}) exceeded"
super().__init__(message, code="MAX_ATTEMPTS", details=details)
# CSV Exceptions
class CSVError(DeckBuilderError):
"""Base exception class for CSV-related errors.
This exception serves as the base for all CSV-related errors in the deck builder,
including file reading, processing, validation, and timeout issues.
Attributes:
code (str): Error code for identifying the error type
message (str): Descriptive error message
details (dict): Additional error context and details
"""
def __init__(self, message: str, code: str = "CSV_ERR", details: dict | None = None):
"""Initialize the base CSV error.
Args:
message: Human-readable error description
code: Error code for identification and handling
details: Additional context about the error
"""
super().__init__(message, code=code, details=details)
class CSVReadError(CSVError):
"""Raised when there are issues reading CSV files.
This exception is used when CSV files cannot be opened, read, or parsed.
"""
def __init__(self, filename: str, details: dict | None = None):
"""Initialize CSV read error.
Args:
filename: Name of the CSV file that failed to read
details: Additional context about the read failure
"""
message = f"Failed to read CSV file: '{filename}'"
super().__init__(message, code="CSV_READ", details=details)
class CSVProcessingError(CSVError):
"""Base class for CSV and DataFrame processing errors.
This exception is used when operations fail during data processing,
including batch operations and transformations.
"""
def __init__(self, message: str, operation_context: dict | None = None, details: dict | None = None):
"""Initialize processing error with context.
Args:
message: Descriptive error message
operation_context: Details about the failed operation
details: Additional error context
"""
if operation_context:
details = details or {}
details['operation_context'] = operation_context
super().__init__(message, code="CSV_PROC", details=details)
class DataFrameProcessingError(CSVProcessingError):
"""Raised when DataFrame batch operations fail.
This exception provides detailed context about batch processing failures
including operation state and progress information.
"""
def __init__(self, operation: str, batch_state: dict, processed_count: int, total_count: int, details: dict | None = None):
"""Initialize DataFrame processing error.
Args:
operation: Name of the operation that failed
batch_state: Current state of batch processing
processed_count: Number of items processed
total_count: Total number of items to process
details: Additional error context
"""
message = f"DataFrame batch operation '{operation}' failed after processing {processed_count}/{total_count} items"
operation_context = {
'operation': operation,
'batch_state': batch_state,
'processed_count': processed_count,
'total_count': total_count
}
super().__init__(message, operation_context, details)
class ColorFilterError(MTGSetupError):
"""Exception raised when color-specific filtering operations fail.
@ -84,113 +257,330 @@ class ColorFilterError(MTGSetupError):
error_info = f" - {details}" if details else ""
super().__init__(f"{message} for color '{color}'{error_info}")
class CommanderValidationError(MTGSetupError):
"""Exception raised when commander validation fails.
class CSVValidationError(CSVError):
"""Base class for CSV and DataFrame validation errors.
This exception is raised when there are issues validating commander cards,
such as non-legendary creatures, color identity mismatches, or banned cards.
Args:
message: Explanation of the error
validation_type: Type of validation that failed (e.g., 'legendary_check', 'color_identity', 'banned_set')
details: Additional error details
Examples:
>>> raise CommanderValidationError(
... "Card must be legendary",
... "legendary_check",
... "Lightning Bolt is not a legendary creature"
... )
>>> raise CommanderValidationError(
... "Commander color identity mismatch",
... "color_identity",
... "Omnath, Locus of Creation cannot be used in Golgari deck"
... )
>>> raise CommanderValidationError(
... "Commander banned in format",
... "banned_set",
... "Golos, Tireless Pilgrim is banned in Commander"
... )
This exception is used when data fails validation checks, including field validation,
data type validation, and data consistency validation.
"""
def __init__(self, message: str, validation_type: str, details: str = None) -> None:
self.validation_type = validation_type
self.details = details
error_info = f" - {details}" if details else ""
super().__init__(f"{message} [{validation_type}]{error_info}")
class InputValidationError(MTGSetupError):
"""Exception raised when input validation fails.
This exception is raised when there are issues validating user input,
such as invalid text formats, number ranges, or confirmation responses.
Args:
message: Explanation of the error
input_type: Type of input validation that failed (e.g., 'text', 'number', 'confirm')
details: Additional error details
Examples:
>>> raise InputValidationError(
... "Invalid number input",
... "number",
... "Value must be between 1 and 100"
... )
def __init__(self, message: str, validation_context: dict | None = None, details: dict | None = None):
"""Initialize validation error with context.
>>> raise InputValidationError(
... "Invalid confirmation response",
... "confirm",
... "Please enter 'y' or 'n'"
... )
>>> raise InputValidationError(
... "Invalid text format",
... "text",
... "Input contains invalid characters"
... )
Args:
message: Descriptive error message
validation_context: Specific validation failure details
details: Additional error context
"""
if validation_context:
details = details or {}
details['validation_context'] = validation_context
super().__init__(message, code="CSV_VALID", details=details)
class DataFrameValidationError(CSVValidationError):
"""Raised when DataFrame validation fails.
This exception provides detailed context about validation failures including
rule violations, invalid values, and data type mismatches.
"""
def __init__(self, message: str, input_type: str, details: str = None) -> None:
self.input_type = input_type
self.details = details
error_info = f" - {details}" if details else ""
super().__init__(f"{message} [{input_type}]{error_info}")
class PriceCheckError(MTGSetupError):
"""Exception raised when price checking operations fail.
This exception is raised when there are issues retrieving or processing
card prices, such as API failures, invalid responses, or parsing errors.
Args:
message: Explanation of the error
card_name: Name of the card that caused the error
details: Additional error details
Examples:
>>> raise PriceCheckError(
... "Failed to retrieve price",
... "Black Lotus",
... "API request timeout"
... )
def __init__(self, field: str, validation_rules: dict, invalid_data: dict | None = None, details: dict | None = None):
"""Initialize DataFrame validation error.
>>> raise PriceCheckError(
... "Invalid price data format",
... "Lightning Bolt",
... "Unexpected response structure"
... )
>>> raise PriceCheckError(
... "Price data unavailable",
... "Underground Sea",
... "No price information found"
... )
Args:
field: Name of the field that failed validation
validation_rules: Rules that were violated
invalid_data: The invalid data that caused the failure
details: Additional error context
"""
message = f"DataFrame validation failed for field '{field}'"
validation_context = {
'field': field,
'rules': validation_rules,
'invalid_data': invalid_data or {}
}
super().__init__(message, validation_context, details)
class EmptyDataFrameError(CSVError):
"""Raised when a DataFrame is unexpectedly empty.
This exception is used when a DataFrame operation requires non-empty data
but receives an empty DataFrame.
"""
def __init__(self, message: str, card_name: str, details: str = None) -> None:
self.card_name = card_name
self.details = details
error_info = f" - {details}" if details else ""
super().__init__(f"{message} for card '{card_name}'{error_info}")
def __init__(self, operation: str, details: dict | None = None):
"""Initialize empty DataFrame error.
Args:
operation: Name of the operation that requires non-empty data
details: Additional context about the empty DataFrame
"""
message = f"Empty DataFrame encountered during: '{operation}'"
super().__init__(message, code="CSV_EMPTY", details=details)
class CSVTimeoutError(CSVError):
"""Base class for CSV and DataFrame timeout errors.
This exception is used when operations exceed their timeout thresholds.
"""
def __init__(self, message: str, timeout_context: dict | None = None, details: dict | None = None):
"""Initialize timeout error with context.
Args:
message: Descriptive error message
timeout_context: Details about the timeout
details: Additional error context
"""
if timeout_context:
details = details or {}
details['timeout_context'] = timeout_context
super().__init__(message, code="CSV_TIMEOUT", details=details)
class DataFrameTimeoutError(CSVTimeoutError):
"""Raised when DataFrame operations timeout.
This exception provides detailed context about operation timeouts
including operation type and duration information.
"""
def __init__(self, operation: str, timeout: float, elapsed: float, operation_state: dict | None = None, details: dict | None = None):
"""Initialize DataFrame timeout error.
Args:
operation: Name of the operation that timed out
timeout: Timeout threshold in seconds
elapsed: Actual time elapsed in seconds
operation_state: State of the operation when timeout occurred
details: Additional error context
"""
message = f"DataFrame operation '{operation}' timed out after {elapsed:.1f}s (threshold: {timeout}s)"
timeout_context = {
'operation': operation,
'timeout_threshold': timeout,
'elapsed_time': elapsed,
'operation_state': operation_state or {}
}
super().__init__(message, timeout_context, details)
# For PriceCheck/Scrython functions
class PriceError(DeckBuilderError):
"""Base exception class for price-related errors.
This exception serves as the base for all price-related errors in the deck builder,
including API issues, validation errors, and price limit violations.
Attributes:
code (str): Error code for identifying the error type
message (str): Descriptive error message
details (dict): Additional error context and details
"""
def __init__(self, message: str, code: str = "PRICE_ERR", details: dict | None = None):
"""Initialize the base price error.
Args:
message: Human-readable error description
code: Error code for identification and handling
details: Additional context about the error
"""
super().__init__(message, code=code, details=details)
class PriceAPIError(PriceError):
"""Raised when there are issues with the Scryfall API price lookup.
This exception is used when the price API request fails, returns invalid data,
or encounters other API-related issues.
"""
def __init__(self, card_name: str, details: dict | None = None):
"""Initialize price API error.
Args:
card_name: Name of the card that failed price lookup
details: Additional context about the API failure
"""
message = f"Failed to fetch price data for '{card_name}' from Scryfall API"
super().__init__(message, code="PRICE_API", details=details)
class PriceLimitError(PriceError):
"""Raised when a card or deck price exceeds the specified limit.
This exception is used when price thresholds are violated during deck building.
"""
def __init__(self, card_name: str, price: float, limit: float, details: dict | None = None):
"""Initialize price limit error.
Args:
card_name: Name of the card exceeding the price limit
price: Actual price of the card
limit: Maximum allowed price
details: Additional context about the price limit violation
"""
message = f"Price of '{card_name}' (${price:.2f}) exceeds limit of ${limit:.2f}"
super().__init__(message, code="PRICE_LIMIT", details=details)
class PriceTimeoutError(PriceError):
"""Raised when a price lookup request times out.
This exception is used when the Scryfall API request exceeds the timeout threshold.
"""
def __init__(self, card_name: str, timeout: float, details: dict | None = None):
"""Initialize price timeout error.
Args:
card_name: Name of the card that timed out
timeout: Timeout threshold in seconds
details: Additional context about the timeout
"""
message = f"Price lookup for '{card_name}' timed out after {timeout} seconds"
super().__init__(message, code="PRICE_TIMEOUT", details=details)
class PriceValidationError(PriceError):
"""Raised when price data fails validation.
This exception is used when received price data is invalid, malformed,
or cannot be properly parsed.
"""
def __init__(self, card_name: str, price_data: str, details: dict | None = None):
"""Initialize price validation error.
Args:
card_name: Name of the card with invalid price data
price_data: The invalid price data received
details: Additional context about the validation failure
"""
message = f"Invalid price data for '{card_name}': {price_data}"
super().__init__(message, code="PRICE_INVALID", details=details)
# Commander Exceptions
class CommanderLoadError(DeckBuilderError):
"""Raised when there are issues loading commander data from CSV.
This exception is used when the commander CSV file cannot be loaded,
is missing required columns, or contains invalid data.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander load error.
Args:
message: Description of the loading failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_LOAD", details=details)
class CommanderSelectionError(DeckBuilderError):
"""Raised when there are issues with the commander selection process.
This exception is used when the commander selection process fails,
such as no matches found or ambiguous matches.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander selection error.
Args:
message: Description of the selection failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_SELECT", details=details)
class CommanderValidationError(DeckBuilderError):
"""Raised when commander data fails validation.
This exception is used when the selected commander's data is invalid,
missing required fields, or contains inconsistent information.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander validation error.
Args:
message: Description of the validation failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_VALID", details=details)
class CommanderTypeError(CommanderValidationError):
"""Raised when commander type validation fails.
This exception is used when a commander fails the legendary creature requirement
or has an invalid creature type.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander type error.
Args:
message: Description of the type validation failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_TYPE_ERR", details=details)
class CommanderStatsError(CommanderValidationError):
"""Raised when commander stats validation fails.
This exception is used when a commander's power, toughness, or mana value
fails validation requirements.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander stats error.
Args:
message: Description of the stats validation failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_STATS_ERR", details=details)
class CommanderColorError(CommanderValidationError):
"""Raised when commander color identity validation fails.
This exception is used when a commander's color identity is invalid
or incompatible with deck requirements.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander color error.
Args:
message: Description of the color validation failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_COLOR_ERR", details=details)
class CommanderTagError(CommanderValidationError):
"""Raised when commander theme tag validation fails.
This exception is used when a commander's theme tags are invalid
or incompatible with deck requirements.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander tag error.
Args:
message: Description of the tag validation failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_TAG_ERR", details=details)
class CommanderThemeError(CommanderValidationError):
"""Raised when commander theme validation fails.
This exception is used when a commander's themes are invalid
or incompatible with deck requirements.
"""
def __init__(self, message: str, details: dict | None = None):
"""Initialize commander theme error.
Args:
message: Description of the theme validation failure
details: Additional context about the error
"""
super().__init__(message, code="CMD_THEME_ERR", details=details)

View file

@ -1,40 +1,74 @@
"""Input validation and handling for MTG Python Deckbuilder.
"""Input handling and validation module for MTG Python Deckbuilder."""
This module provides the InputHandler class which encapsulates all input validation
and handling logic. It supports different types of input validation including text,
numbers, confirmations, and multiple choice questions.
"""
from __future__ import annotations
from typing import Any, List, Optional, Union
import inquirer
import logging
import os
from typing import Any, List, Optional, Tuple, Union
from exceptions import InputValidationError
from settings import INPUT_VALIDATION, QUESTION_TYPES
import inquirer.prompt # type: ignore
from settings import (
COLORS, COLOR_ABRV, DEFAULT_MAX_CARD_PRICE,
DEFAULT_MAX_DECK_PRICE, DEFAULT_THEME_TAGS, MONO_COLOR_MAP,
DUAL_COLOR_MAP, TRI_COLOR_MAP, OTHER_COLOR_MAP
)
# Create logs directory if it doesn't exist
if not os.path.exists('logs'):
os.makedirs('logs')
from exceptions import (
CommanderColorError,
CommanderStatsError,
CommanderTagError,
CommanderThemeError,
CommanderTypeError,
DeckBuilderError,
EmptyInputError,
InvalidNumberError,
InvalidQuestionTypeError,
MaxAttemptsError,
PriceError,
PriceLimitError,
PriceValidationError
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler('logs/input_handlers.log', mode='a', encoding='utf-8')
]
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class InputHandler:
"""Handles input validation and user interaction.
"""Handles user input operations with validation and error handling.
This class provides methods for validating different types of user input
and handling user interaction through questionnaires. It uses constants
from settings.py for validation messages and configuration.
This class provides methods for collecting and validating different types
of user input including text, numbers, confirmations, and choices.
Attributes:
max_attempts (int): Maximum number of retry attempts for invalid input
default_text (str): Default value for text input
default_number (float): Default value for number input
default_confirm (bool): Default value for confirmation input
"""
def __init__(
self,
max_attempts: int = 3,
default_text: str = '',
default_number: float = 0.0,
default_confirm: bool = True
):
"""Initialize input handler with configuration.
Args:
max_attempts: Maximum number of retry attempts
default_text: Default value for text input
default_number: Default value for number input
default_confirm: Default value for confirmation input
"""
self.max_attempts = max_attempts
self.default_text = default_text
self.default_number = default_number
self.default_confirm = default_confirm
def validate_text(self, result: str) -> bool:
"""Validate text input is not empty.
@ -42,171 +76,367 @@ class InputHandler:
result: Text input to validate
Returns:
bool: True if text is not empty after stripping whitespace
True if text is not empty after stripping whitespace
Raises:
InputValidationError: If text validation fails
EmptyInputError: If input is empty or whitespace only
"""
try:
if not result or not result.strip():
raise InputValidationError(
INPUT_VALIDATION['default_text_message'],
'text',
'Input cannot be empty'
)
return True
except Exception as e:
raise InputValidationError(
str(e),
'text',
'Unexpected error during text validation'
)
def validate_number(self, result: str) -> Optional[float]:
if not result or not result.strip():
raise EmptyInputError()
return True
def validate_number(self, result: str) -> float:
"""Validate and convert string input to float.
Args:
result: Number input to validate
Returns:
float | None: Converted float value or None if invalid
Converted float value
Raises:
InputValidationError: If number validation fails
InvalidNumberError: If input cannot be converted to float
"""
try:
if not result:
raise InputValidationError(
INPUT_VALIDATION['default_number_message'],
'number',
'Input cannot be empty'
)
return float(result)
except ValueError:
raise InputValidationError(
INPUT_VALIDATION['default_number_message'],
'number',
'Input must be a valid number'
)
except Exception as e:
raise InputValidationError(
str(e),
'number',
'Unexpected error during number validation'
)
except (ValueError, TypeError):
raise InvalidNumberError(result)
def validate_confirm(self, result: Any) -> bool:
def validate_price(self, result: str) -> Tuple[float, bool]:
"""Validate and convert price input to float with format checking.
Args:
result: Price input to validate
Returns:
Tuple of (price value, is_unlimited flag)
Raises:
PriceValidationError: If price format is invalid
"""
result = result.strip().lower()
# Check for unlimited budget
if result in ['unlimited', 'any']:
return (float('inf'), True)
# Remove currency symbol if present
if result.startswith('$'):
result = result[1:]
try:
price = float(result)
if price < 0:
raise PriceValidationError('Price cannot be negative')
return (price, False)
except ValueError:
raise PriceValidationError(f"Invalid price format: '{result}'")
def validate_price_threshold(self, price: float, threshold: float = DEFAULT_MAX_CARD_PRICE) -> bool:
"""Validate price against maximum threshold.
Args:
price: Price value to check
threshold: Maximum allowed price (default from settings)
Returns:
True if price is within threshold
Raises:
PriceLimitError: If price exceeds threshold
"""
if price > threshold and price != float('inf'):
raise PriceLimitError('Card', price, threshold)
return True
def validate_confirm(self, result: bool) -> bool:
"""Validate confirmation input.
Args:
result: Confirmation input to validate
result: Boolean confirmation input
Returns:
bool: True for positive confirmation, False otherwise
Raises:
InputValidationError: If confirmation validation fails
The boolean input value
"""
try:
if isinstance(result, bool):
return result
if isinstance(result, str):
result = result.lower().strip()
if result in ('y', 'yes', 'true', '1'):
return True
if result in ('n', 'no', 'false', '0'):
return False
raise InputValidationError(
INPUT_VALIDATION['default_confirm_message'],
'confirm',
'Invalid confirmation response'
)
except InputValidationError:
raise
except Exception as e:
raise InputValidationError(
str(e),
'confirm',
'Unexpected error during confirmation validation'
)
return bool(result)
def questionnaire(
self,
question_type: str,
default_value: Union[str, bool, float] = '',
choices_list: List[str] = []
) -> Union[str, bool, float]:
"""Present questions to user and validate input.
message: str = '',
default_value: Any = None,
choices_list: List[str] = None
) -> Union[str, float, bool]:
"""Present questions to user and handle input validation.
Args:
question_type: Type of question ('Text', 'Number', 'Confirm', 'Choice')
message: Question message to display
default_value: Default value for the question
choices_list: List of choices for Choice type questions
Returns:
Union[str, bool, float]: Validated user input
Validated user input of appropriate type
Raises:
InputValidationError: If input validation fails
ValueError: If question type is not supported
InvalidQuestionTypeError: If question_type is not supported
MaxAttemptsError: If maximum retry attempts are exceeded
"""
if question_type not in QUESTION_TYPES:
raise ValueError(f"Unsupported question type: {question_type}")
attempts = 0
while attempts < INPUT_VALIDATION['max_attempts']:
while attempts < self.max_attempts:
try:
if question_type == 'Text':
question = [inquirer.Text('text')]
question = [
inquirer.Text(
'text',
message=message or 'Enter text',
default=default_value or self.default_text
)
]
result = inquirer.prompt(question)['text']
if self.validate_text(result):
return result
elif question_type == 'Price':
question = [
inquirer.Text(
'price',
message=message or 'Enter price (or "unlimited")',
default=str(default_value or DEFAULT_MAX_CARD_PRICE)
)
]
result = inquirer.prompt(question)['price']
price, is_unlimited = self.validate_price(result)
if not is_unlimited:
self.validate_price_threshold(price)
return price
elif question_type == 'Number':
question = [inquirer.Text('number', default=str(default_value))]
question = [
inquirer.Text(
'number',
message=message or 'Enter number',
default=str(default_value or self.default_number)
)
]
result = inquirer.prompt(question)['number']
validated = self.validate_number(result)
if validated is not None:
return validated
return self.validate_number(result)
elif question_type == 'Confirm':
question = [inquirer.Confirm('confirm', default=default_value)]
question = [
inquirer.Confirm(
'confirm',
message=message or 'Confirm?',
default=default_value if default_value is not None else self.default_confirm
)
]
result = inquirer.prompt(question)['confirm']
return self.validate_confirm(result)
elif question_type == 'Choice':
if not choices_list:
raise InputValidationError(
INPUT_VALIDATION['default_choice_message'],
'choice',
'No choices provided'
)
raise ValueError("Choices list cannot be empty for Choice type")
question = [
inquirer.List('selection',
inquirer.List(
'selection',
message=message or 'Select an option',
choices=choices_list,
carousel=True)
carousel=True
)
]
return inquirer.prompt(question)['selection']
except InputValidationError as e:
else:
raise InvalidQuestionTypeError(question_type)
except DeckBuilderError as e:
logger.warning(f"Input validation failed: {e}")
attempts += 1
if attempts >= INPUT_VALIDATION['max_attempts']:
raise InputValidationError(
"Maximum input attempts reached",
question_type,
str(e)
if attempts >= self.max_attempts:
raise MaxAttemptsError(
self.max_attempts,
question_type.lower(),
{"last_error": str(e)}
)
logger.warning(f"Invalid input ({attempts}/{INPUT_VALIDATION['max_attempts']}): {str(e)}")
except Exception as e:
raise InputValidationError(
str(e),
question_type,
'Unexpected error during questionnaire'
logger.error(f"Unexpected error in questionnaire: {e}")
raise
raise MaxAttemptsError(self.max_attempts, question_type.lower())
def validate_commander_type(self, type_line: str) -> str:
"""Validate commander type line requirements.
Args:
type_line: Commander's type line to validate
Returns:
Validated type line
Raises:
CommanderTypeError: If type line validation fails
"""
if not type_line:
raise CommanderTypeError("Type line cannot be empty")
type_line = type_line.strip()
# Check for legendary creature requirement
if not ('Legendary' in type_line and 'Creature' in type_line):
# Check for 'can be your commander' text
if 'can be your commander' not in type_line.lower():
raise CommanderTypeError(
"Commander must be a legendary creature or have 'can be your commander' text"
)
raise InputValidationError(
"Maximum input attempts reached",
question_type,
"Failed to get valid input"
)
return type_line
def validate_commander_stats(self, stat_name: str, value: str) -> int:
"""Validate commander numerical statistics.
Args:
stat_name: Name of the stat (power, toughness, mana value)
value: Value to validate
Returns:
Validated integer value
Raises:
CommanderStatsError: If stat validation fails
"""
try:
stat_value = int(value)
if stat_value < 0 and stat_name != 'power':
raise CommanderStatsError(f"{stat_name} cannot be negative")
return stat_value
except ValueError:
raise CommanderStatsError(
f"Invalid {stat_name} value: '{value}'. Must be a number."
)
def _normalize_color_string(self, colors: str) -> str:
"""Helper method to standardize color string format.
Args:
colors: Raw color string to normalize
Returns:
Normalized color string
"""
if not colors:
return 'colorless'
# Remove whitespace and sort color symbols
colors = colors.strip().upper()
color_symbols = [c for c in colors if c in 'WUBRG']
return ', '.join(sorted(color_symbols))
def _validate_color_combination(self, colors: str) -> bool:
"""Helper method to validate color combinations.
Args:
colors: Normalized color string to validate
Returns:
True if valid, False otherwise
"""
if colors == 'colorless':
return True
# Check against valid combinations from settings
return (colors in COLOR_ABRV or
any(colors in combo for combo in [MONO_COLOR_MAP, DUAL_COLOR_MAP,
TRI_COLOR_MAP, OTHER_COLOR_MAP]))
def validate_color_identity(self, colors: str) -> str:
"""Validate commander color identity using settings constants.
Args:
colors: Color identity string to validate
Returns:
Validated color identity string
Raises:
CommanderColorError: If color validation fails
"""
# Normalize the color string
normalized = self._normalize_color_string(colors)
# Validate the combination
if not self._validate_color_combination(normalized):
raise CommanderColorError(
f"Invalid color identity: '{colors}'. Must be a valid color combination."
)
return normalized
def validate_commander_colors(self, colors: str) -> str:
"""Validate commander color identity.
Args:
colors: Color identity string to validate
Returns:
Validated color identity string
Raises:
CommanderColorError: If color validation fails
"""
try:
return self.validate_color_identity(colors)
except CommanderColorError as e:
logger.error(f"Color validation failed: {e}")
raise
def validate_commander_tags(self, tags: List[str]) -> List[str]:
"""Validate commander theme tags.
Args:
tags: List of theme tags to validate
Returns:
Validated list of theme tags
Raises:
CommanderTagError: If tag validation fails
"""
if not isinstance(tags, list):
raise CommanderTagError("Tags must be provided as a list")
validated_tags = []
for tag in tags:
if not isinstance(tag, str):
raise CommanderTagError(f"Invalid tag type: {type(tag)}. Must be string.")
tag = tag.strip()
if tag:
validated_tags.append(tag)
return validated_tags
def validate_commander_themes(self, themes: List[str]) -> List[str]:
"""Validate commander themes.
Args:
themes: List of themes to validate
Returns:
Validated list of themes
Raises:
CommanderThemeError: If theme validation fails
"""
if not isinstance(themes, list):
raise CommanderThemeError("Themes must be provided as a list")
validated_themes = []
for theme in themes:
if not isinstance(theme, str):
raise CommanderThemeError(f"Invalid theme type: {type(theme)}. Must be string.")
theme = theme.strip()
if theme and theme in DEFAULT_THEME_TAGS:
validated_themes.append(theme)
else:
raise CommanderThemeError(f"Invalid theme: '{theme}'")
return validated_themes

View file

@ -4,57 +4,204 @@ This module provides functionality to check card prices using the Scryfall API
through the scrython library. It includes caching and error handling for reliable
price lookups.
"""
from __future__ import annotations
# Standard library imports
import logging
import time
from functools import lru_cache
from typing import Optional
from typing import Dict, List, Optional, Tuple, Union
# Third-party imports
import scrython
from scrython.cards import Named
from exceptions import PriceCheckError
from settings import PRICE_CHECK_CONFIG
@lru_cache(maxsize=PRICE_CHECK_CONFIG['cache_size'])
def check_price(card_name: str) -> float:
"""Retrieve the current price of a Magic: The Gathering card.
Args:
card_name: The name of the card to check.
Returns:
float: The current price of the card in USD.
Raises:
PriceCheckError: If there are any issues retrieving the price.
from scrython.cards import Named as ScryfallCard
# Local imports
from exceptions import (
PriceAPIError,
PriceLimitError,
PriceTimeoutError,
PriceValidationError
)
from settings import (
BATCH_PRICE_CHECK_SIZE,
DEFAULT_MAX_CARD_PRICE,
DEFAULT_MAX_DECK_PRICE,
DEFAULT_PRICE_DELAY,
MAX_PRICE_CHECK_ATTEMPTS,
PRICE_CACHE_SIZE,
PRICE_CHECK_TIMEOUT,
PRICE_TOLERANCE_MULTIPLIER
)
from type_definitions import PriceCache
class PriceChecker:
"""Class for handling MTG card price checking and validation.
This class provides functionality for checking card prices via the Scryfall API,
validating prices against thresholds, and managing price caching.
Attributes:
price_cache (Dict[str, float]): Cache of card prices
max_card_price (float): Maximum allowed price per card
max_deck_price (float): Maximum allowed total deck price
current_deck_price (float): Current total price of the deck
"""
retries = 0
last_error = None
while retries < PRICE_CHECK_CONFIG['max_retries']:
def __init__(
self,
max_card_price: float = DEFAULT_MAX_CARD_PRICE,
max_deck_price: float = DEFAULT_MAX_DECK_PRICE
) -> None:
"""Initialize the PriceChecker.
Args:
max_card_price: Maximum allowed price per card
max_deck_price: Maximum allowed total deck price
"""
self.price_cache: PriceCache = {}
self.max_card_price: float = max_card_price
self.max_deck_price: float = max_deck_price
self.current_deck_price: float = 0.0
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
@lru_cache(maxsize=PRICE_CACHE_SIZE)
def get_card_price(self, card_name: str, attempts: int = 0) -> float:
"""Get the price of a card with caching and retry logic.
Args:
card_name: Name of the card to check
attempts: Current number of retry attempts
Returns:
Float price of the card in USD
Raises:
PriceAPIError: If price lookup fails after max attempts
PriceTimeoutError: If request times out
PriceValidationError: If received price data is invalid
"""
# Check cache first
if card_name in self.price_cache:
return self.price_cache[card_name]
try:
card = Named(fuzzy=card_name)
price = card.prices('usd')
print(price)
# Add delay between API calls
time.sleep(DEFAULT_PRICE_DELAY)
if price is None:
raise PriceCheckError(
"No price data available",
card_name,
"Card may be too new or not available in USD"
)
# Make API request with type hints
card: ScryfallCard = scrython.cards.Named(fuzzy=card_name, timeout=PRICE_CHECK_TIMEOUT)
price: Optional[str] = card.prices('usd')
return float(price)
except (scrython.ScryfallError, ValueError) as e:
last_error = str(e)
retries += 1
if retries < PRICE_CHECK_CONFIG['max_retries']:
time.sleep(0.1) # Brief delay before retry
continue
raise PriceCheckError(
"Failed to retrieve price after multiple attempts",
card_name,
f"Last error: {last_error}"
)
# Handle None or empty string cases
if price is None or price == "":
return 0.0
# Validate and cache price
if isinstance(price, (int, float, str)):
try:
# Convert string or numeric price to float
price_float = float(price)
self.price_cache[card_name] = price_float
return price_float
except ValueError:
raise PriceValidationError(card_name, str(price))
return 0.0
except scrython.foundation.ScryfallError as e:
if attempts < MAX_PRICE_CHECK_ATTEMPTS:
logging.warning(f"Retrying price check for {card_name} (attempt {attempts + 1})")
return self.get_card_price(card_name, attempts + 1)
raise PriceAPIError(card_name, {"error": str(e)})
except TimeoutError:
raise PriceTimeoutError(card_name, PRICE_CHECK_TIMEOUT)
except Exception as e:
if attempts < MAX_PRICE_CHECK_ATTEMPTS:
logging.warning(f"Unexpected error checking price for {card_name}, retrying")
return self.get_card_price(card_name, attempts + 1)
raise PriceAPIError(card_name, {"error": str(e)})
def validate_card_price(self, card_name: str, price: float) -> bool | None:
"""Validate if a card's price is within allowed limits.
Args:
card_name: Name of the card to validate
price: Price to validate
Returns:
True if price is valid, False otherwise
Raises:
PriceLimitError: If price exceeds maximum allowed
"""
if price > self.max_card_price * PRICE_TOLERANCE_MULTIPLIER:
raise PriceLimitError(card_name, price, self.max_card_price)
return True
def validate_deck_price(self) -> bool | None:
"""Validate if the current deck price is within allowed limits.
Returns:
True if deck price is valid, False otherwise
Raises:
PriceLimitError: If deck price exceeds maximum allowed
"""
if self.current_deck_price > self.max_deck_price * PRICE_TOLERANCE_MULTIPLIER:
raise PriceLimitError("deck", self.current_deck_price, self.max_deck_price)
return True
def batch_check_prices(self, card_names: List[str]) -> Dict[str, float]:
"""Check prices for multiple cards efficiently.
Args:
card_names: List of card names to check prices for
Returns:
Dictionary mapping card names to their prices
Raises:
PriceAPIError: If batch price lookup fails
"""
results: Dict[str, float] = {}
errors: List[Tuple[str, Exception]] = []
# Process in batches
for i in range(0, len(card_names), BATCH_PRICE_CHECK_SIZE):
batch = card_names[i:i + BATCH_PRICE_CHECK_SIZE]
for card_name in batch:
try:
price = self.get_card_price(card_name)
results[card_name] = price
except Exception as e:
errors.append((card_name, e))
logging.error(f"Error checking price for {card_name}: {e}")
if errors:
logging.warning(f"Failed to get prices for {len(errors)} cards")
return results
def update_deck_price(self, price: float) -> None:
"""Update the current deck price.
Args:
price: Price to add to current deck total
"""
self.current_deck_price += price
logging.debug(f"Updated deck price to ${self.current_deck_price:.2f}")
def clear_cache(self) -> None:
"""Clear the price cache."""
self.price_cache.clear()
self.get_card_price.cache_clear()
logging.info("Price cache cleared")

View file

@ -1,16 +1,44 @@
"""Constants and configuration settings for the MTG Python Deckbuilder.
from typing import Dict, List, Optional, Final, Tuple, Pattern, Union
import ast
This module contains all the constant values and configuration settings used throughout
the application for card filtering, processing, and analysis. Constants are organized
into logical sections with clear documentation.
# Commander selection configuration
COMMANDER_CSV_PATH: Final[str] = 'csv_files/commander_cards.csv'
FUZZY_MATCH_THRESHOLD: Final[int] = 90 # Threshold for fuzzy name matching
MAX_FUZZY_CHOICES: Final[int] = 5 # Maximum number of fuzzy match choices
COMMANDER_CONVERTERS: Final[Dict[str, str]] = {'themeTags': ast.literal_eval, 'creatureTypes': ast.literal_eval} # CSV loading converters
All constants are properly typed according to PEP 484 standards to ensure type safety
and enable static type checking with mypy.
"""
from typing import Dict, List, Optional
# Commander-related constants
COMMANDER_POWER_DEFAULT: Final[int] = 0
COMMANDER_TOUGHNESS_DEFAULT: Final[int] = 0
COMMANDER_MANA_VALUE_DEFAULT: Final[int] = 0
COMMANDER_TYPE_DEFAULT: Final[str] = ''
COMMANDER_TEXT_DEFAULT: Final[str] = ''
COMMANDER_MANA_COST_DEFAULT: Final[str] = ''
COMMANDER_COLOR_IDENTITY_DEFAULT: Final[str] = ''
COMMANDER_COLORS_DEFAULT: Final[List[str]] = []
COMMANDER_CREATURE_TYPES_DEFAULT: Final[str] = ''
COMMANDER_TAGS_DEFAULT: Final[List[str]] = []
COMMANDER_THEMES_DEFAULT: Final[List[str]] = []
# Price checking configuration
DEFAULT_PRICE_DELAY: Final[float] = 0.1 # Delay between price checks in seconds
MAX_PRICE_CHECK_ATTEMPTS: Final[int] = 3 # Maximum attempts for price checking
PRICE_CACHE_SIZE: Final[int] = 128 # Size of price check LRU cache
PRICE_CHECK_TIMEOUT: Final[int] = 30 # Timeout for price check requests in seconds
PRICE_TOLERANCE_MULTIPLIER: Final[float] = 1.1 # Multiplier for price tolerance
DEFAULT_MAX_CARD_PRICE: Final[float] = 20.0 # Default maximum price per card
DEFAULT_MAX_DECK_PRICE: Final[float] = 400.0 # Default maximum total deck price
BATCH_PRICE_CHECK_SIZE: Final[int] = 50 # Number of cards to check prices for in one batch
# Constants for input validation
# Type aliases
CardName = str
CardType = str
ThemeTag = str
ColorIdentity = str
ColorList = List[str]
ColorInfo = Tuple[str, List[str], List[str]]
INPUT_VALIDATION = {
'max_attempts': 3,
'default_text_message': 'Please enter a valid text response.',
@ -52,8 +80,7 @@ banned_cards = [# in commander
'Jihad', 'Imprison', 'Crusade'
]
basic_lands = ['Plains', 'Island', 'Swamp', 'Mountain', 'Forest']
basic_lands = ['Plains', 'Island', 'Swamp', 'Mountain', 'Forest']
BASIC_LANDS = ['Plains', 'Island', 'Swamp', 'Mountain', 'Forest']
# Constants for lands matter functionality
LANDS_MATTER_PATTERNS: Dict[str, List[str]] = {
@ -530,7 +557,7 @@ BOARD_WIPE_EXCLUSION_PATTERNS = [
]
card_types = ['Artifact','Creature', 'Enchantment', 'Instant', 'Land', 'Planeswalker', 'Sorcery',
CARD_TYPES = ['Artifact','Creature', 'Enchantment', 'Instant', 'Land', 'Planeswalker', 'Sorcery',
'Kindred', 'Dungeon', 'Battle']
# Mapping of card types to their corresponding theme tags
@ -547,9 +574,102 @@ TYPE_TAG_MAPPING = {
'Sorcery': ['Spells Matter', 'Spellslinger']
}
csv_directory = 'csv_files'
CSV_DIRECTORY = 'csv_files'
colors = ['colorless', 'white', 'blue', 'black', 'red', 'green',
# Color identity constants and mappings
MONO_COLOR_MAP: Final[Dict[str, Tuple[str, List[str]]]] = {
'COLORLESS': ('Colorless', ['colorless']),
'W': ('White', ['colorless', 'white']),
'U': ('Blue', ['colorless', 'blue']),
'B': ('Black', ['colorless', 'black']),
'R': ('Red', ['colorless', 'red']),
'G': ('Green', ['colorless', 'green'])
}
DUAL_COLOR_MAP: Final[Dict[str, Tuple[str, List[str], List[str]]]] = {
'B, G': ('Golgari: Black/Green', ['B', 'G', 'B, G'], ['colorless', 'black', 'green', 'golgari']),
'B, R': ('Rakdos: Black/Red', ['B', 'R', 'B, R'], ['colorless', 'black', 'red', 'rakdos']),
'B, U': ('Dimir: Black/Blue', ['B', 'U', 'B, U'], ['colorless', 'black', 'blue', 'dimir']),
'B, W': ('Orzhov: Black/White', ['B', 'W', 'B, W'], ['colorless', 'black', 'white', 'orzhov']),
'G, R': ('Gruul: Green/Red', ['G', 'R', 'G, R'], ['colorless', 'green', 'red', 'gruul']),
'G, U': ('Simic: Green/Blue', ['G', 'U', 'G, U'], ['colorless', 'green', 'blue', 'simic']),
'G, W': ('Selesnya: Green/White', ['G', 'W', 'G, W'], ['colorless', 'green', 'white', 'selesnya']),
'R, U': ('Izzet: Blue/Red', ['U', 'R', 'U, R'], ['colorless', 'blue', 'red', 'izzet']),
'U, W': ('Azorius: Blue/White', ['U', 'W', 'U, W'], ['colorless', 'blue', 'white', 'azorius']),
'R, W': ('Boros: Red/White', ['R', 'W', 'R, W'], ['colorless', 'red', 'white', 'boros'])
}
TRI_COLOR_MAP: Final[Dict[str, Tuple[str, List[str], List[str]]]] = {
'B, G, U': ('Sultai: Black/Blue/Green', ['B', 'G', 'U', 'B, G', 'B, U', 'G, U', 'B, G, U'],
['colorless', 'black', 'blue', 'green', 'dimir', 'golgari', 'simic', 'sultai']),
'B, G, R': ('Jund: Black/Red/Green', ['B', 'G', 'R', 'B, G', 'B, R', 'G, R', 'B, G, R'],
['colorless', 'black', 'green', 'red', 'golgari', 'rakdos', 'gruul', 'jund']),
'B, G, W': ('Abzan: Black/Green/White', ['B', 'G', 'W', 'B, G', 'B, W', 'G, W', 'B, G, W'],
['colorless', 'black', 'green', 'white', 'golgari', 'orzhov', 'selesnya', 'abzan']),
'B, R, U': ('Grixis: Black/Blue/Red', ['B', 'R', 'U', 'B, R', 'B, U', 'R, U', 'B, R, U'],
['colorless', 'black', 'blue', 'red', 'dimir', 'rakdos', 'izzet', 'grixis']),
'B, R, W': ('Mardu: Black/Red/White', ['B', 'R', 'W', 'B, R', 'B, W', 'R, W', 'B, R, W'],
['colorless', 'black', 'red', 'white', 'rakdos', 'orzhov', 'boros', 'mardu']),
'B, U, W': ('Esper: Black/Blue/White', ['B', 'U', 'W', 'B, U', 'B, W', 'U, W', 'B, U, W'],
['colorless', 'black', 'blue', 'white', 'dimir', 'orzhov', 'azorius', 'esper']),
'G, R, U': ('Temur: Blue/Green/Red', ['G', 'R', 'U', 'G, R', 'G, U', 'R, U', 'G, R, U'],
['colorless', 'green', 'red', 'blue', 'simic', 'izzet', 'gruul', 'temur']),
'G, R, W': ('Naya: Green/Red/White', ['G', 'R', 'W', 'G, R', 'G, W', 'R, W', 'G, R, W'],
['colorless', 'green', 'red', 'white', 'gruul', 'selesnya', 'boros', 'naya']),
'G, U, W': ('Bant: Blue/Green/White', ['G', 'U', 'W', 'G, U', 'G, W', 'U, W', 'G, U, W'],
['colorless', 'green', 'blue', 'white', 'simic', 'azorius', 'selesnya', 'bant']),
'R, U, W': ('Jeskai: Blue/Red/White', ['R', 'U', 'W', 'R, U', 'U, W', 'R, W', 'R, U, W'],
['colorless', 'blue', 'red', 'white', 'izzet', 'azorius', 'boros', 'jeskai'])
}
OTHER_COLOR_MAP: Final[Dict[str, Tuple[str, List[str], List[str]]]] = {
'B, G, R, U': ('Glint: Black/Blue/Green/Red',
['B', 'G', 'R', 'U', 'B, G', 'B, R', 'B, U', 'G, R', 'G, U', 'R, U', 'B, G, R',
'B, G, U', 'B, R, U', 'G, R, U', 'B, G, R, U'],
['colorless', 'black', 'blue', 'green', 'red', 'golgari', 'rakdos', 'dimir',
'gruul', 'simic', 'izzet', 'jund', 'sultai', 'grixis', 'temur', 'glint']),
'B, G, R, W': ('Dune: Black/Green/Red/White',
['B', 'G', 'R', 'W', 'B, G', 'B, R', 'B, W', 'G, R', 'G, W', 'R, W', 'B, G, R',
'B, G, W', 'B, R, W', 'G, R, W', 'B, G, R, W'],
['colorless', 'black', 'green', 'red', 'white', 'golgari', 'rakdos', 'orzhov',
'gruul', 'selesnya', 'boros', 'jund', 'abzan', 'mardu', 'naya', 'dune']),
'B, G, U, W': ('Witch: Black/Blue/Green/White',
['B', 'G', 'U', 'W', 'B, G', 'B, U', 'B, W', 'G, U', 'G, W', 'U, W', 'B, G, U',
'B, G, W', 'B, U, W', 'G, U, W', 'B, G, U, W'],
['colorless', 'black', 'blue', 'green', 'white', 'golgari', 'dimir', 'orzhov',
'simic', 'selesnya', 'azorius', 'sultai', 'abzan', 'esper', 'bant', 'witch']),
'B, R, U, W': ('Yore: Black/Blue/Red/White',
['B', 'R', 'U', 'W', 'B, R', 'B, U', 'B, W', 'R, U', 'R, W', 'U, W', 'B, R, U',
'B, R, W', 'B, U, W', 'R, U, W', 'B, R, U, W'],
['colorless', 'black', 'blue', 'red', 'white', 'rakdos', 'dimir', 'orzhov',
'izzet', 'boros', 'azorius', 'grixis', 'mardu', 'esper', 'jeskai', 'yore']),
'G, R, U, W': ('Ink: Blue/Green/Red/White',
['G', 'R', 'U', 'W', 'G, R', 'G, U', 'G, W', 'R, U', 'R, W', 'U, W', 'G, R, U',
'G, R, W', 'G, U, W', 'R, U, W', 'G, R, U, W'],
['colorless', 'blue', 'green', 'red', 'white', 'gruul', 'simic', 'selesnya',
'izzet', 'boros', 'azorius', 'temur', 'naya', 'bant', 'jeskai', 'ink']),
'B, G, R, U, W': ('WUBRG: All colors',
['B', 'G', 'R', 'U', 'W', 'B, G', 'B, R', 'B, U', 'B, W', 'G, R', 'G, U',
'G, W', 'R, U', 'R, W', 'U, W', 'B, G, R', 'B, G, U', 'B, G, W', 'B, R, U',
'B, R, W', 'B, U, W', 'G, R, U', 'G, R, W', 'G, U, W', 'R, U, W',
'B, G, R, U', 'B, G, R, W', 'B, G, U, W', 'B, R, U, W', 'G, R, U, W',
'B, G, R, U, W'],
['colorless', 'black', 'green', 'red', 'blue', 'white', 'golgari', 'rakdos',
'dimir', 'orzhov', 'gruul', 'simic', 'selesnya', 'izzet', 'boros', 'azorius',
'jund', 'sultai', 'abzan', 'grixis', 'mardu', 'esper', 'temur', 'naya',
'bant', 'jeskai', 'glint', 'dune', 'witch', 'yore', 'ink', 'wubrg'])
}
# Color identity validation patterns
COLOR_IDENTITY_PATTERNS: Final[Dict[str, str]] = {
'mono': r'^[WUBRG]$',
'dual': r'^[WUBRG], [WUBRG]$',
'tri': r'^[WUBRG], [WUBRG], [WUBRG]$',
'four': r'^[WUBRG], [WUBRG], [WUBRG], [WUBRG]$',
'five': r'^[WUBRG], [WUBRG], [WUBRG], [WUBRG], [WUBRG]$'
}
COLORS = ['colorless', 'white', 'blue', 'black', 'red', 'green',
'azorius', 'orzhov', 'selesnya', 'boros', 'dimir',
'simic', 'izzet', 'golgari', 'rakdos', 'gruul',
'bant', 'esper', 'grixis', 'jund', 'naya',
@ -707,6 +827,21 @@ COLUMN_ORDER = [
'power', 'toughness', 'keywords', 'themeTags', 'layout', 'side'
]
PRETAG_COLUMN_ORDER: List[str] = [
'name', 'faceName', 'edhrecRank', 'colorIdentity', 'colors',
'manaCost', 'manaValue', 'type', 'text', 'power', 'toughness',
'keywords', 'layout', 'side'
]
TAGGED_COLUMN_ORDER: List[str] = [
'name', 'faceName', 'edhrecRank', 'colorIdentity', 'colors',
'manaCost', 'manaValue', 'type', 'creatureTypes', 'text',
'power', 'toughness', 'keywords', 'themeTags', 'layout', 'side'
]
EXCLUDED_CARD_TYPES: List[str] = ['Plane —', 'Conspiracy', 'Vanguard', 'Scheme',
'Phenomenon', 'Stickers', 'Attraction', 'Hero',
'Contraption']
# Constants for type detection and processing
OUTLAW_TYPES = ['Assassin', 'Mercenary', 'Pirate', 'Rogue', 'Warlock']
TYPE_DETECTION_BATCH_SIZE = 1000
@ -757,7 +892,7 @@ EQUIPMENT_TEXT_PATTERNS = [
'unattach', # Equipment removal
'unequip', # Equipment removal
]
TYPE_DETECTION_BATCH_SIZE = 1000
# Constants for Voltron strategy
VOLTRON_COMMANDER_CARDS = [
@ -808,6 +943,101 @@ PRICE_CHECK_CONFIG: Dict[str, float] = {
# Price tolerance factor (e.g., 1.1 means accept prices within 10% difference)
'price_tolerance': 1.1
}
# DataFrame processing configuration
BATCH_SIZE: Final[int] = 1000 # Number of records to process at once
DATAFRAME_BATCH_SIZE: Final[int] = 500 # Batch size for DataFrame operations
TRANSFORM_BATCH_SIZE: Final[int] = 250 # Batch size for data transformations
CSV_DOWNLOAD_TIMEOUT: Final[int] = 30 # Timeout in seconds for CSV downloads
PROGRESS_UPDATE_INTERVAL: Final[int] = 100 # Number of records between progress updates
# DataFrame operation timeouts
DATAFRAME_READ_TIMEOUT: Final[int] = 30 # Timeout for DataFrame read operations
DATAFRAME_WRITE_TIMEOUT: Final[int] = 30 # Timeout for DataFrame write operations
DATAFRAME_TRANSFORM_TIMEOUT: Final[int] = 45 # Timeout for DataFrame transformations
DATAFRAME_VALIDATION_TIMEOUT: Final[int] = 20 # Timeout for DataFrame validation
# DataFrame validation configuration
MIN_EDHREC_RANK: int = 0
MAX_EDHREC_RANK: int = 100000
MIN_MANA_VALUE: int = 0
MAX_MANA_VALUE: int = 20
# DataFrame validation rules
DATAFRAME_VALIDATION_RULES: Final[Dict[str, Dict[str, Union[str, int, float, bool]]]] = {
'name': {'type': ('str', 'object'), 'required': True, 'unique': True},
'edhrecRank': {'type': ('str', 'int', 'float', 'object'), 'min': 0, 'max': 100000},
'manaValue': {'type': ('str', 'int', 'float', 'object'), 'min': 0, 'max': 20},
'power': {'type': ('str', 'int', 'float', 'object'), 'pattern': r'^[\d*+-]+$'},
'toughness': {'type': ('str', 'int', 'float', 'object'), 'pattern': r'^[\d*+-]+$'},
'colorIdentity': {'type': ('str', 'object'), 'required': True},
'text': {'type': ('str', 'object'), 'required': False}
}
# Card category validation rules
CREATURE_VALIDATION_RULES: Final[Dict[str, Dict[str, Union[str, int, float, bool]]]] = {
'power': {'type': ('str', 'int', 'float'), 'required': True},
'toughness': {'type': ('str', 'int', 'float'), 'required': True},
'creatureTypes': {'type': 'list', 'required': True}
}
SPELL_VALIDATION_RULES: Final[Dict[str, Dict[str, Union[str, int, float, bool]]]] = {
'manaCost': {'type': 'str', 'required': True},
'text': {'type': 'str', 'required': True}
}
LAND_VALIDATION_RULES: Final[Dict[str, Dict[str, Union[str, int, float, bool]]]] = {
'type': {'type': ('str', 'object'), 'required': True},
'text': {'type': ('str', 'object'), 'required': False}
}
# Column mapping configurations
DATAFRAME_COLUMN_MAPS: Final[Dict[str, Dict[str, str]]] = {
'creature': {
'name': 'Card Name',
'type': 'Card Type',
'manaCost': 'Mana Cost',
'manaValue': 'Mana Value',
'power': 'Power',
'toughness': 'Toughness'
},
'spell': {
'name': 'Card Name',
'type': 'Card Type',
'manaCost': 'Mana Cost',
'manaValue': 'Mana Value'
},
'land': {
'name': 'Card Name',
'type': 'Card Type'
}
}
# Required DataFrame columns
DATAFRAME_REQUIRED_COLUMNS: Final[List[str]] = [
'name', 'type', 'colorIdentity', 'manaValue', 'text',
'edhrecRank', 'themeTags', 'keywords'
]
# CSV processing configuration
CSV_READ_TIMEOUT: Final[int] = 30 # Timeout in seconds for CSV read operations
CSV_PROCESSING_BATCH_SIZE: Final[int] = 1000 # Number of rows to process in each batch
# CSV validation configuration
CSV_VALIDATION_RULES: Final[Dict[str, Dict[str, Union[str, int, float]]]] = {
'name': {'type': ('str', 'object'), 'required': True, 'unique': True},
'edhrecRank': {'type': ('str', 'int', 'float', 'object'), 'min': 0, 'max': 100000},
'manaValue': {'type': ('str', 'int', 'float', 'object'), 'min': 0, 'max': 20},
'power': {'type': ('str', 'int', 'float', 'object'), 'pattern': r'^[\d*+-]+$'},
'toughness': {'type': ('str', 'int', 'float', 'object'), 'pattern': r'^[\d*+-]+$'}
}
# Required columns for CSV validation
CSV_REQUIRED_COLUMNS: Final[List[str]] = [
'name', 'faceName', 'edhrecRank', 'colorIdentity', 'colors',
'manaCost', 'manaValue', 'type', 'creatureTypes', 'text',
'power', 'toughness', 'keywords', 'themeTags', 'layout', 'side'
]
# Constants for setup and CSV processing
MTGJSON_API_URL = 'https://mtgjson.com/api/v5/csv/cards.csv'

View file

@ -13,7 +13,7 @@ import inquirer
# Local application imports
from settings import (
banned_cards, csv_directory, SETUP_COLORS, COLOR_ABRV, MTGJSON_API_URL
banned_cards, CSV_DIRECTORY, SETUP_COLORS, COLOR_ABRV, MTGJSON_API_URL
)
from setup_utils import (
download_cards_csv, filter_dataframe, process_legendary_cards, filter_by_color_identity
@ -72,7 +72,7 @@ def initial_setup() -> None:
logger.info('Checking for cards.csv file')
try:
cards_file = f'{csv_directory}/cards.csv'
cards_file = f'{CSV_DIRECTORY}/cards.csv'
try:
with open(cards_file, 'r', encoding='utf-8'):
logger.info('cards.csv exists')
@ -88,11 +88,11 @@ def initial_setup() -> None:
for i in range(min(len(SETUP_COLORS), len(COLOR_ABRV))):
logger.info(f'Checking for {SETUP_COLORS[i]}_cards.csv')
try:
with open(f'{csv_directory}/{SETUP_COLORS[i]}_cards.csv', 'r', encoding='utf-8'):
with open(f'{CSV_DIRECTORY}/{SETUP_COLORS[i]}_cards.csv', 'r', encoding='utf-8'):
logger.info(f'{SETUP_COLORS[i]}_cards.csv exists')
except FileNotFoundError:
logger.info(f'{SETUP_COLORS[i]}_cards.csv not found, creating one')
filter_by_color(df, 'colorIdentity', COLOR_ABRV[i], f'{csv_directory}/{SETUP_COLORS[i]}_cards.csv')
filter_by_color(df, 'colorIdentity', COLOR_ABRV[i], f'{CSV_DIRECTORY}/{SETUP_COLORS[i]}_cards.csv')
# Generate commander list
determine_commanders()
@ -143,7 +143,7 @@ def determine_commanders() -> None:
try:
# Check for cards.csv with progress tracking
cards_file = f'{csv_directory}/cards.csv'
cards_file = f'{CSV_DIRECTORY}/cards.csv'
if not check_csv_exists(cards_file):
logger.info('cards.csv not found, initiating download')
download_cards_csv(MTGJSON_API_URL, cards_file)
@ -169,7 +169,7 @@ def determine_commanders() -> None:
# Save commander cards
logger.info('Saving validated commander cards')
filtered_df.to_csv(f'{csv_directory}/commander_cards.csv', index=False)
filtered_df.to_csv(f'{CSV_DIRECTORY}/commander_cards.csv', index=False)
logger.info('Commander card generation completed successfully')
@ -196,10 +196,10 @@ def regenerate_csvs_all() -> None:
"""
try:
logger.info('Downloading latest card data from MTGJSON')
download_cards_csv(MTGJSON_API_URL, f'{csv_directory}/cards.csv')
download_cards_csv(MTGJSON_API_URL, f'{CSV_DIRECTORY}/cards.csv')
logger.info('Loading and processing card data')
df = pd.read_csv(f'{csv_directory}/cards.csv', low_memory=False)
df = pd.read_csv(f'{CSV_DIRECTORY}/cards.csv', low_memory=False)
df['colorIdentity'] = df['colorIdentity'].fillna('Colorless')
logger.info('Regenerating color identity sorted files')
@ -207,7 +207,7 @@ def regenerate_csvs_all() -> None:
color = SETUP_COLORS[i]
color_id = COLOR_ABRV[i]
logger.info(f'Processing {color} cards')
filter_by_color(df, 'colorIdentity', color_id, f'{csv_directory}/{color}_cards.csv')
filter_by_color(df, 'colorIdentity', color_id, f'{CSV_DIRECTORY}/{color}_cards.csv')
logger.info('Regenerating commander cards')
determine_commanders()
@ -239,14 +239,14 @@ def regenerate_csv_by_color(color: str) -> None:
color_abv = COLOR_ABRV[SETUP_COLORS.index(color)]
logger.info(f'Downloading latest card data for {color} cards')
download_cards_csv(MTGJSON_API_URL, f'{csv_directory}/cards.csv')
download_cards_csv(MTGJSON_API_URL, f'{CSV_DIRECTORY}/cards.csv')
logger.info('Loading and processing card data')
df = pd.read_csv(f'{csv_directory}/cards.csv', low_memory=False)
df = pd.read_csv(f'{CSV_DIRECTORY}/cards.csv', low_memory=False)
df['colorIdentity'] = df['colorIdentity'].fillna('Colorless')
logger.info(f'Regenerating {color} cards CSV')
filter_by_color(df, 'colorIdentity', color_abv, f'{csv_directory}/{color}_cards.csv')
filter_by_color(df, 'colorIdentity', color_abv, f'{CSV_DIRECTORY}/{color}_cards.csv')
logger.info(f'Successfully regenerated {color} cards database')

View file

@ -36,6 +36,10 @@ from settings import (
FILL_NA_COLUMNS,
SORT_CONFIG,
FILTER_CONFIG,
COLUMN_ORDER,
PRETAG_COLUMN_ORDER,
EXCLUDED_CARD_TYPES,
TAGGED_COLUMN_ORDER
)
from exceptions import (
MTGJSONDownloadError,
@ -43,6 +47,7 @@ from exceptions import (
ColorFilterError,
CommanderValidationError
)
from type_definitions import CardLibraryDF
# Create logs directory if it doesn't exist
if not os.path.exists('logs'):
@ -339,4 +344,74 @@ def process_legendary_cards(df: pd.DataFrame) -> pd.DataFrame:
"Failed to process legendary cards",
"commander_processing",
str(e)
) from e
) from e
def process_card_dataframe(df: CardLibraryDF, batch_size: int = 1000, columns_to_keep: Optional[List[str]] = None,
include_commander_cols: bool = False, skip_availability_checks: bool = False) -> pd.DataFrame:
"""Process DataFrame with common operations in batches.
Args:
df: DataFrame to process
batch_size: Size of batches for processing
columns_to_keep: List of columns to keep (default: COLUMN_ORDER)
include_commander_cols: Whether to include commander-specific columns
skip_availability_checks: Whether to skip availability and security checks (default: False)
Args:
df: DataFrame to process
batch_size: Size of batches for processing
columns_to_keep: List of columns to keep (default: COLUMN_ORDER)
include_commander_cols: Whether to include commander-specific columns
Returns:
CardLibraryDF: Processed DataFrame with standardized structure
"""
logger.info("Processing card DataFrame...")
if columns_to_keep is None:
columns_to_keep = TAGGED_COLUMN_ORDER.copy()
if include_commander_cols:
commander_cols = ['printings', 'text', 'power', 'toughness', 'keywords']
columns_to_keep.extend(col for col in commander_cols if col not in columns_to_keep)
# Fill NA values
df.loc[:, 'colorIdentity'] = df['colorIdentity'].fillna('Colorless')
df.loc[:, 'faceName'] = df['faceName'].fillna(df['name'])
# Process in batches
total_batches = len(df) // batch_size + 1
processed_dfs = []
for i in tqdm(range(total_batches), desc="Processing batches"):
start_idx = i * batch_size
end_idx = min((i + 1) * batch_size, len(df))
batch = df.iloc[start_idx:end_idx].copy()
if not skip_availability_checks:
columns_to_keep = COLUMN_ORDER.copy()
logger.debug("Performing column checks...")
# Common processing steps
batch = batch[batch['availability'].str.contains('paper', na=False)]
batch = batch.loc[batch['layout'] != 'reversible_card']
batch = batch.loc[batch['promoTypes'] != 'playtest']
batch = batch.loc[batch['securityStamp'] != 'heart']
batch = batch.loc[batch['securityStamp'] != 'acorn']
# Keep only specified columns
batch = batch[columns_to_keep]
processed_dfs.append(batch)
else:
logger.debug("Skipping column checks...")
# Keep only specified columns
batch = batch[columns_to_keep]
processed_dfs.append(batch)
# Combine processed batches
result = pd.concat(processed_dfs, ignore_index=True)
# Final processing
result.drop_duplicates(subset='faceName', keep='first', inplace=True)
result.sort_values(by=['name', 'side'], key=lambda col: col.str.lower(), inplace=True)
logger.info("DataFrame processing completed")
return result

View file

@ -13,7 +13,7 @@ import settings # type: ignore
import tag_utils # type: ignore
# Local application imports
from settings import csv_directory, multiple_copy_cards, num_to_search, triggers
from settings import CSV_DIRECTORY, multiple_copy_cards, num_to_search, triggers
from setup import regenerate_csv_by_color
@ -68,7 +68,7 @@ def load_dataframe(color: str) -> None:
ValueError: If required columns are missing
"""
try:
filepath = f'{csv_directory}/{color}_cards.csv'
filepath = f'{CSV_DIRECTORY}/{color}_cards.csv'
# Check if file exists, regenerate if needed
if not os.path.exists(filepath):
@ -170,7 +170,7 @@ def tag_by_color(df: pd.DataFrame, color: str) -> None:
# Lastly, sort all theme tags for easier reading
sort_theme_tags(df, color)
df.to_csv(f'{csv_directory}/{color}_cards.csv', index=False)
df.to_csv(f'{CSV_DIRECTORY}/{color}_cards.csv', index=False)
#print(df)
print('\n====================\n')
logger.info(f'Tags are done being set on {color}_cards.csv')
@ -255,7 +255,7 @@ def kindred_tagging(df: pd.DataFrame, color: str) -> None:
'keywords', 'layout', 'side'
]
df = df[columns_to_keep]
df.to_csv(f'{settings.csv_directory}/{color}_cards.csv', index=False)
df.to_csv(f'{settings.CSV_DIRECTORY}/{color}_cards.csv', index=False)
total_time = pd.Timestamp.now() - start_time
logger.info(f'Creature type tagging completed in {total_time.total_seconds():.2f}s')
@ -320,7 +320,7 @@ def create_theme_tags(df: pd.DataFrame, color: str) -> None:
# Save results
try:
df.to_csv(f'{settings.csv_directory}/{color}_cards.csv', index=False)
df.to_csv(f'{settings.CSV_DIRECTORY}/{color}_cards.csv', index=False)
total_time = pd.Timestamp.now() - start_time
logger.info(f'Creature type tagging completed in {total_time.total_seconds():.2f}s')
@ -6425,6 +6425,4 @@ def run_tagging():
for color in settings.colors:
load_dataframe(color)
duration = (pd.Timestamp.now() - start_time).total_seconds()
logger.info(f'Tagged cards in {duration:.2f}s')
run_tagging()
logger.info(f'Tagged cards in {duration:.2f}s')

49
type_definitions.py Normal file
View file

@ -0,0 +1,49 @@
from __future__ import annotations
from typing import Dict, List, TypedDict, Union
import pandas as pd
class CardDict(TypedDict):
"""Type definition for card dictionary structure used in deck_builder.py.
Contains all the necessary fields to represent a Magic: The Gathering card
in the deck building process.
"""
name: str
type: str
manaCost: Union[str, None]
manaValue: int
class CommanderDict(TypedDict):
"""Type definition for commander dictionary structure used in deck_builder.py.
Contains all the necessary fields to represent a commander card and its
associated metadata.
"""
Commander_Name: str
Mana_Cost: str
Mana_Value: int
Color_Identity: str
Colors: List[str]
Type: str
Creature_Types: str
Text: str
Power: int
Toughness: int
Themes: List[str]
CMC: float
# Type alias for price cache dictionary used in price_checker.py
PriceCache = Dict[str, float]
# DataFrame type aliases for different card categories
CardLibraryDF = pd.DataFrame
CommanderDF = pd.DataFrame
LandDF = pd.DataFrame
ArtifactDF = pd.DataFrame
CreatureDF = pd.DataFrame
NonCreatureDF = pd.DataFrame
EnchantmentDF = pd.DataFrame
InstantDF = pd.DataFrame
PlaneswalkerDF = pd.DataFrame
SorceryDF = pd.DataFrame