300 lines
11 KiB
Python
300 lines
11 KiB
Python
import json
|
|
import random
|
|
from typing import Any, Dict, List, Optional
|
|
from sqlmodel import Session, select
|
|
from .models import Game, Player, GameEvent
|
|
from . import cards
|
|
|
|
FACE_VALUES = ("J", "Q", "K")
|
|
|
|
# --- Card and Hand Utilities ---
|
|
|
|
def get_player_hand(player: Player) -> List[str]:
|
|
return json.loads(player.hand_cards)
|
|
|
|
def set_player_hand(player: Player, hand: List[str]):
|
|
player.hand_cards = json.dumps(hand)
|
|
|
|
def get_game_deck(game: Game) -> List[str]:
|
|
return json.loads(game.deck_cards)
|
|
|
|
def set_game_deck(game: Game, deck: List[str]):
|
|
game.deck_cards = json.dumps(deck)
|
|
|
|
def technique_for_face_card(player: Player, value: str) -> str:
|
|
"""The Secret Pirate Technique a player has assigned to a face card value (J/Q/K)."""
|
|
return {
|
|
"J": player.tech_jack or "Jack Secret Technique",
|
|
"Q": player.tech_queen or "Queen Secret Technique",
|
|
"K": player.tech_king or "King Secret Technique",
|
|
}[value]
|
|
|
|
def evaluate_card_play(
|
|
player: Player,
|
|
card_code: str,
|
|
obstacle_card_code: str,
|
|
obstacle_value: int,
|
|
acting_rank: int,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Shared Challenge resolution rules for one (non-Joker) card played by a Pi-Rat
|
|
(callers must have already rejected Jokers and Deep players):
|
|
- J/Q/K triggers the player's assigned Secret Pirate Technique: automatic success.
|
|
- A face card acting as the Obstacle is worth the challenged Pi-Rat's Rank
|
|
(acting_rank); any other Obstacle card is worth obstacle_value.
|
|
- Gat privilege: Black cards +1. Name privilege: Red cards +1.
|
|
Returns {"success", "details", "is_technique", "tech_name"}.
|
|
"""
|
|
played = cards.parse_card(card_code)
|
|
|
|
if played["value"] in FACE_VALUES:
|
|
tech_name = technique_for_face_card(player, played["value"])
|
|
return {
|
|
"success": True,
|
|
"details": f"Used Secret Pirate Technique: '{tech_name}'!",
|
|
"is_technique": True,
|
|
"tech_name": tech_name,
|
|
}
|
|
|
|
if cards.parse_card(obstacle_card_code)["value"] in FACE_VALUES:
|
|
target_value = acting_rank
|
|
obs_detail = f"Rank {acting_rank} (Face Card Obstacle)"
|
|
else:
|
|
target_value = obstacle_value
|
|
obs_detail = str(target_value)
|
|
|
|
privilege_bonus = 0
|
|
if player.completed_personal_1 and played["suit"] in ("C", "S"):
|
|
privilege_bonus += 1
|
|
if player.completed_personal_2 and played["suit"] in ("H", "D"):
|
|
privilege_bonus += 1
|
|
final_value = played["numeric_value"] + privilege_bonus
|
|
|
|
success = final_value > target_value
|
|
outcome = "Success" if success else "Failure"
|
|
return {
|
|
"success": success,
|
|
"details": f"{outcome}! Played {played['display']} ({final_value}) vs Obstacle value {obs_detail}.",
|
|
"is_technique": False,
|
|
"tech_name": "",
|
|
}
|
|
|
|
def calculate_max_hand_size(player: Player, players_in_scene: List[Player], captain_id: Optional[str] = None) -> int:
|
|
"""
|
|
Calculates a player's maximum hand size based on their Rank relative to others,
|
|
plus Captain privileges (+1, regardless of Rank).
|
|
Rank Hand sizes:
|
|
- Highest Rank Pi-Rat(s): max 4 cards
|
|
- Middle Rank Pi-Rat(s): max 3 cards
|
|
- Lowest Rank Pi-Rat(s): max 2 cards
|
|
If everyone shares a Rank, they are all 'highest' and get 4 cards.
|
|
Pi-Rats are ranked against the Pi-Rats in the scene; Deep players (whose hand size
|
|
only matters for the between-scenes redraw) are ranked against all players.
|
|
"""
|
|
if player.role == "deep":
|
|
pool = players_in_scene
|
|
else:
|
|
pool = [p for p in players_in_scene if p.role != "deep"]
|
|
if not pool:
|
|
pool = [player]
|
|
|
|
ranks = [p.rank for p in pool]
|
|
if player.rank >= max(ranks):
|
|
base_size = 4
|
|
elif player.rank <= min(ranks):
|
|
base_size = 2
|
|
else:
|
|
base_size = 3
|
|
|
|
if captain_id is not None and player.id == captain_id:
|
|
return base_size + 1
|
|
return base_size
|
|
|
|
def is_player_captain(player: Player, game: Game) -> bool:
|
|
"""The Captain is a persistent position tracked on the Game (set once the crew steals a ship)."""
|
|
return game.captain_player_id is not None and game.captain_player_id == player.id
|
|
|
|
def capture_hand_maxes(players: List[Player], captain_id: Optional[str] = None) -> dict:
|
|
"""Snapshot every player's max hand size, for detecting increases after a Rank/Captain change."""
|
|
return {p.id: calculate_max_hand_size(p, players, captain_id) for p in players}
|
|
|
|
def apply_hand_increases(db: Session, game: Game, old_maxes: dict):
|
|
"""
|
|
'Anytime your Hand size increases, you immediately draw a card.'
|
|
Compares current max hand sizes against a snapshot and draws the difference
|
|
for any player whose max increased. (Decreases never force a discard.)
|
|
"""
|
|
for p in game.players:
|
|
new_max = calculate_max_hand_size(p, game.players, game.captain_player_id)
|
|
old_max = old_maxes.get(p.id, new_max)
|
|
if new_max > old_max:
|
|
draw_cards_for_player(db, game, p, new_max - old_max)
|
|
|
|
def change_player_rank(db: Session, game: Game, player: Player, delta: int):
|
|
"""Changes a player's Rank and grants immediate draws for any resulting hand size increases."""
|
|
old_maxes = capture_hand_maxes(game.players, game.captain_player_id)
|
|
player.rank = max(1, player.rank + delta)
|
|
db.add(player)
|
|
db.commit()
|
|
apply_hand_increases(db, game, old_maxes)
|
|
|
|
def set_captain(db: Session, game: Game, player_id: Optional[str], reason: str = ""):
|
|
"""Assigns (or clears) the Captain, granting the +1 hand size draw to the new Captain."""
|
|
if game.captain_player_id == player_id:
|
|
return
|
|
old_maxes = capture_hand_maxes(game.players, game.captain_player_id)
|
|
game.captain_player_id = player_id
|
|
db.add(game)
|
|
db.commit()
|
|
apply_hand_increases(db, game, old_maxes)
|
|
if player_id:
|
|
new_captain = db.get(Player, player_id)
|
|
if new_captain:
|
|
msg = f"{new_captain.name} is now the Captain!"
|
|
if reason:
|
|
msg += f" ({reason})"
|
|
add_game_event(db, game.id, msg, kind="captain")
|
|
elif reason:
|
|
add_game_event(db, game.id, f"The Captain's position is vacant. ({reason})", kind="captain")
|
|
|
|
def reshuffle_discard_pile(db: Session, game: Game):
|
|
"""
|
|
Gathers all cards not currently in any player's hand and not active on the table,
|
|
adds them back to the deck, and shuffles them.
|
|
"""
|
|
# 1. Get all cards in players' hands
|
|
all_hand_cards = set()
|
|
for player in game.players:
|
|
all_hand_cards.update(get_player_hand(player))
|
|
|
|
# 2. Get active cards on the table
|
|
# Obstacle cards and their whole played-card columns stay on the table until the Obstacle is discarded
|
|
active_table_cards = set()
|
|
for obs in game.obstacles:
|
|
active_table_cards.add(obs.original_card)
|
|
played = json.loads(obs.played_cards)
|
|
for entry in played:
|
|
active_table_cards.add(entry["card"])
|
|
# A PvP challenger's temporary Obstacle is on the table until the duel resolves
|
|
for challenge in game.challenges:
|
|
if challenge.status == "open" and challenge.temp_card:
|
|
active_table_cards.add(challenge.temp_card)
|
|
|
|
# 3. The full deck of 54 cards
|
|
full_deck = []
|
|
for suit in cards.SUITS.keys():
|
|
for val in cards.VALUES:
|
|
full_deck.append(f"{val}{suit}")
|
|
full_deck.extend(["Joker1", "Joker2"])
|
|
|
|
# 4. Remaining cards to shuffle
|
|
remaining_cards = [c for c in full_deck if c not in all_hand_cards and c not in active_table_cards]
|
|
random.shuffle(remaining_cards)
|
|
|
|
set_game_deck(game, remaining_cards)
|
|
db.add(game)
|
|
db.commit()
|
|
|
|
def draw_cards_for_player(db: Session, game: Game, player: Player, count: int) -> List[str]:
|
|
"""Draws count cards from the deck for a player, reshuffling if necessary."""
|
|
deck = get_game_deck(game)
|
|
hand = get_player_hand(player)
|
|
drawn = []
|
|
|
|
for _ in range(count):
|
|
if not deck:
|
|
# Reshuffle discard pile
|
|
reshuffle_discard_pile(db, game)
|
|
deck = get_game_deck(game)
|
|
if not deck:
|
|
# If still empty (all 54 cards are in hands or active), we can't draw
|
|
break
|
|
card = deck.pop(0)
|
|
hand.append(card)
|
|
drawn.append(card)
|
|
|
|
set_game_deck(game, deck)
|
|
set_player_hand(player, hand)
|
|
db.add(game)
|
|
db.add(player)
|
|
db.commit()
|
|
return drawn
|
|
|
|
# --- CRUD Operations ---
|
|
|
|
def create_game(db: Session, crew_name: Optional[str] = None) -> Game:
|
|
deck = cards.get_fresh_deck()
|
|
game = Game(
|
|
deck_cards=json.dumps(deck),
|
|
phase="lobby",
|
|
current_scene_number=1,
|
|
extra_obstacles=0,
|
|
crew_name=crew_name
|
|
)
|
|
db.add(game)
|
|
db.commit()
|
|
db.refresh(game)
|
|
return game
|
|
|
|
def get_game(db: Session, game_id: str) -> Optional[Game]:
|
|
return db.get(Game, game_id)
|
|
|
|
def get_player(db: Session, player_id: str) -> Optional[Player]:
|
|
return db.get(Player, player_id)
|
|
|
|
def add_player(db: Session, game_id: str, name: str, is_creator: bool = False) -> Player:
|
|
game = db.get(Game, game_id)
|
|
|
|
# Players who join after initial character creation create their Pi-Rat in
|
|
# the between-scenes recruit_creation phase; until then they spectate.
|
|
needs_reroll = bool(game and game.phase not in ("lobby", "character_creation"))
|
|
|
|
player = Player(
|
|
game_id=game_id,
|
|
name=name, # the Pi-Rat goes by "Recruit {smell}" once character creation fills in a smell
|
|
player_name=name,
|
|
is_creator=is_creator,
|
|
rank=2, # default starting rank (will be set properly when starting character creation)
|
|
hand_cards="[]",
|
|
is_ready=False,
|
|
role=None,
|
|
needs_reroll=needs_reroll
|
|
)
|
|
db.add(player)
|
|
db.commit()
|
|
db.refresh(player)
|
|
|
|
if game:
|
|
db.refresh(game)
|
|
if game.phase == "character_creation":
|
|
# Late joiner to creation: hand out their Like/Hate questions
|
|
from .crud_character import auto_delegate_all
|
|
auto_delegate_all(db, game)
|
|
elif game.phase == "recruit_creation":
|
|
# Recruit creation is already underway; join it immediately
|
|
from .crud_character import activate_recruit
|
|
activate_recruit(db, game, player)
|
|
|
|
msg = f"{name} joined the crew!"
|
|
if needs_reroll and game and game.phase != "recruit_creation":
|
|
msg += " They'll create their Pi-Rat with the crew between scenes."
|
|
add_game_event(db, game_id, msg, kind="join")
|
|
|
|
return player
|
|
|
|
def add_game_event(db: Session, game_id: str, message: str, kind: str = "info"):
|
|
event = GameEvent(game_id=game_id, message=message, kind=kind)
|
|
db.add(event)
|
|
db.commit()
|
|
|
|
def get_events_page(db: Session, game_id: str, before: Optional[float] = None, limit: int = 50):
|
|
"""Returns the newest `limit` events older than `before` (newest overall if None),
|
|
in ascending timestamp order, plus a flag for whether older events remain."""
|
|
query = select(GameEvent).where(GameEvent.game_id == game_id)
|
|
if before is not None:
|
|
query = query.where(GameEvent.timestamp < before)
|
|
query = query.order_by(GameEvent.timestamp.desc()).limit(limit + 1)
|
|
events = db.exec(query).all()
|
|
has_more = len(events) > limit
|
|
return list(reversed(events[:limit])), has_more
|