437 lines
14 KiB
Python
437 lines
14 KiB
Python
import json
|
||
import random
|
||
import sqlite3
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
from flask import Flask, g, jsonify, render_template, request, session
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
DB_PATH = str(BASE_DIR / "schiffe_versenken.db")
|
||
BOARD_SIZE = 10
|
||
|
||
FLEET = [
|
||
{"id": "koenig", "name": "Oberwappler", "size": 5, "hat": "#c0392b"},
|
||
{"id": "hauptmann", "name": "Chefwappler", "size": 4, "hat": "#2980b9"},
|
||
{"id": "krieger1", "name": "Wappler I", "size": 3, "hat": "#27ae60"},
|
||
{"id": "krieger2", "name": "Wappler II", "size": 3, "hat": "#8e44ad"},
|
||
{"id": "kundschafter", "name": "Wapplertrupp", "size": 2, "hat": "#d35400"},
|
||
]
|
||
FLEET_BY_ID = {s["id"]: s for s in FLEET}
|
||
|
||
app = Flask(__name__)
|
||
app.secret_key = "zwerge-schiffe-versenken-" + uuid.uuid4().hex
|
||
|
||
|
||
def get_db():
|
||
if "db" not in g:
|
||
g.db = sqlite3.connect(DB_PATH)
|
||
g.db.row_factory = sqlite3.Row
|
||
return g.db
|
||
|
||
|
||
@app.teardown_appcontext
|
||
def close_db(exception=None):
|
||
db = g.pop("db", None)
|
||
if db is not None:
|
||
db.close()
|
||
|
||
|
||
def init_db():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.execute(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS games (
|
||
id TEXT PRIMARY KEY,
|
||
player_board TEXT NOT NULL,
|
||
computer_board TEXT NOT NULL,
|
||
player_shots TEXT NOT NULL,
|
||
computer_shots TEXT NOT NULL,
|
||
computer_ai_state TEXT NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'placing',
|
||
message TEXT DEFAULT '',
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
"""
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def empty_grid():
|
||
return [[None for _ in range(BOARD_SIZE)] for _ in range(BOARD_SIZE)]
|
||
|
||
|
||
def empty_shots():
|
||
return [[None for _ in range(BOARD_SIZE)] for _ in range(BOARD_SIZE)]
|
||
|
||
|
||
def can_place(grid, row, col, size, orientation):
|
||
cells = ship_cells(row, col, size, orientation)
|
||
for r, c in cells:
|
||
if not (0 <= r < BOARD_SIZE and 0 <= c < BOARD_SIZE):
|
||
return False
|
||
if grid[r][c] is not None:
|
||
return False
|
||
return True
|
||
|
||
|
||
def ship_cells(row, col, size, orientation):
|
||
if orientation == "h":
|
||
return [(row, col + i) for i in range(size)]
|
||
return [(row + i, col) for i in range(size)]
|
||
|
||
|
||
def place_ship(grid, ship_id, row, col, size, orientation):
|
||
for r, c in ship_cells(row, col, size, orientation):
|
||
grid[r][c] = ship_id
|
||
|
||
|
||
def random_place_all_ships():
|
||
grid = empty_grid()
|
||
for ship in FLEET:
|
||
placed = False
|
||
attempts = 0
|
||
while not placed and attempts < 500:
|
||
attempts += 1
|
||
orientation = random.choice(["h", "v"])
|
||
row = random.randint(0, BOARD_SIZE - 1)
|
||
col = random.randint(0, BOARD_SIZE - 1)
|
||
if can_place(grid, row, col, ship["size"], orientation):
|
||
place_ship(grid, ship["id"], row, col, ship["size"], orientation)
|
||
placed = True
|
||
if not placed:
|
||
return random_place_all_ships()
|
||
return grid
|
||
|
||
|
||
def new_game_row():
|
||
game_id = str(uuid.uuid4())
|
||
computer_board = random_place_all_ships()
|
||
row = {
|
||
"id": game_id,
|
||
"player_board": empty_grid(),
|
||
"computer_board": computer_board,
|
||
"player_shots": empty_shots(),
|
||
"computer_shots": empty_shots(),
|
||
"computer_ai_state": {"mode": "random", "targets": [], "tried": []},
|
||
"status": "placing",
|
||
"message": "Platziere deine Wappler auf dem Spielfeld.",
|
||
}
|
||
db = get_db()
|
||
db.execute(
|
||
"INSERT INTO games (id, player_board, computer_board, player_shots, computer_shots, "
|
||
"computer_ai_state, status, message) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(
|
||
row["id"],
|
||
json.dumps(row["player_board"]),
|
||
json.dumps(row["computer_board"]),
|
||
json.dumps(row["player_shots"]),
|
||
json.dumps(row["computer_shots"]),
|
||
json.dumps(row["computer_ai_state"]),
|
||
row["status"],
|
||
row["message"],
|
||
),
|
||
)
|
||
db.commit()
|
||
session["game_id"] = game_id
|
||
return load_game(game_id)
|
||
|
||
|
||
def load_game(game_id):
|
||
db = get_db()
|
||
r = db.execute("SELECT * FROM games WHERE id = ?", (game_id,)).fetchone()
|
||
if r is None:
|
||
return None
|
||
return {
|
||
"id": r["id"],
|
||
"player_board": json.loads(r["player_board"]),
|
||
"computer_board": json.loads(r["computer_board"]),
|
||
"player_shots": json.loads(r["player_shots"]),
|
||
"computer_shots": json.loads(r["computer_shots"]),
|
||
"computer_ai_state": json.loads(r["computer_ai_state"]),
|
||
"status": r["status"],
|
||
"message": r["message"],
|
||
}
|
||
|
||
|
||
def save_game(game):
|
||
db = get_db()
|
||
db.execute(
|
||
"UPDATE games SET player_board=?, computer_board=?, player_shots=?, computer_shots=?, "
|
||
"computer_ai_state=?, status=?, message=? WHERE id=?",
|
||
(
|
||
json.dumps(game["player_board"]),
|
||
json.dumps(game["computer_board"]),
|
||
json.dumps(game["player_shots"]),
|
||
json.dumps(game["computer_shots"]),
|
||
json.dumps(game["computer_ai_state"]),
|
||
game["status"],
|
||
game["message"],
|
||
game["id"],
|
||
),
|
||
)
|
||
db.commit()
|
||
|
||
|
||
def get_current_game():
|
||
game_id = session.get("game_id")
|
||
if not game_id:
|
||
return None
|
||
return load_game(game_id)
|
||
|
||
|
||
def ship_is_sunk(board, shots, ship_id):
|
||
for r in range(BOARD_SIZE):
|
||
for c in range(BOARD_SIZE):
|
||
if board[r][c] == ship_id and shots[r][c] != "hit":
|
||
return False
|
||
return True
|
||
|
||
|
||
def all_ships_sunk(board, shots):
|
||
ship_ids = {cell for row in board for cell in row if cell is not None}
|
||
return all(ship_is_sunk(board, shots, sid) for sid in ship_ids)
|
||
|
||
|
||
def fleet_status(board, shots):
|
||
result = []
|
||
for ship in FLEET:
|
||
result.append(
|
||
{
|
||
"id": ship["id"],
|
||
"name": ship["name"],
|
||
"size": ship["size"],
|
||
"hat": ship["hat"],
|
||
"sunk": ship_is_sunk(board, shots, ship["id"]),
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def neighbors(r, c):
|
||
for dr, dc in ((1, 0), (-1, 0), (0, 1), (0, -1)):
|
||
nr, nc = r + dr, c + dc
|
||
if 0 <= nr < BOARD_SIZE and 0 <= nc < BOARD_SIZE:
|
||
yield nr, nc
|
||
|
||
|
||
def computer_take_shot(game):
|
||
board = game["player_board"]
|
||
shots = game["computer_shots"]
|
||
ai = game["computer_ai_state"]
|
||
|
||
target = None
|
||
if ai["mode"] == "target" and ai["targets"]:
|
||
while ai["targets"]:
|
||
candidate = tuple(ai["targets"].pop(0))
|
||
if shots[candidate[0]][candidate[1]] is None:
|
||
target = candidate
|
||
break
|
||
if not ai["targets"] and target is None:
|
||
ai["mode"] = "random"
|
||
|
||
if target is None:
|
||
ai["mode"] = "random"
|
||
choices = [
|
||
(r, c)
|
||
for r in range(BOARD_SIZE)
|
||
for c in range(BOARD_SIZE)
|
||
if shots[r][c] is None
|
||
]
|
||
target = random.choice(choices)
|
||
|
||
r, c = target
|
||
hit = board[r][c] is not None
|
||
shots[r][c] = "hit" if hit else "miss"
|
||
|
||
sunk_ship = None
|
||
if hit:
|
||
ship_id = board[r][c]
|
||
if ship_is_sunk(board, shots, ship_id):
|
||
sunk_ship = FLEET_BY_ID[ship_id]["name"]
|
||
ai["mode"] = "random"
|
||
ai["targets"] = []
|
||
else:
|
||
ai["mode"] = "target"
|
||
for nr, nc in neighbors(r, c):
|
||
if shots[nr][nc] is None and [nr, nc] not in ai["targets"]:
|
||
ai["targets"].append([nr, nc])
|
||
game["computer_ai_state"] = ai
|
||
return {"row": r, "col": c, "hit": hit, "sunk_ship": sunk_ship}
|
||
|
||
|
||
@app.route("/")
|
||
def index():
|
||
game = get_current_game()
|
||
if game is None:
|
||
game = new_game_row()
|
||
return render_template("index.html", fleet=FLEET, board_size=BOARD_SIZE)
|
||
|
||
|
||
@app.route("/api/state")
|
||
def api_state():
|
||
game = get_current_game()
|
||
if game is None:
|
||
game = new_game_row()
|
||
return jsonify(build_state_payload(game))
|
||
|
||
|
||
def build_state_payload(game):
|
||
return {
|
||
"status": game["status"],
|
||
"message": game["message"],
|
||
"player_board": game["player_board"],
|
||
"player_shots_received": game["computer_shots"],
|
||
"player_shots_made": game["player_shots"],
|
||
"player_fleet": fleet_status(game["player_board"], game["computer_shots"]),
|
||
"computer_fleet": fleet_status(game["computer_board"], game["player_shots"]),
|
||
"fleet_def": FLEET,
|
||
"board_size": BOARD_SIZE,
|
||
}
|
||
|
||
|
||
@app.route("/api/new_game", methods=["POST"])
|
||
def api_new_game():
|
||
game = new_game_row()
|
||
return jsonify(build_state_payload(game))
|
||
|
||
|
||
@app.route("/api/place_ship", methods=["POST"])
|
||
def api_place_ship():
|
||
game = get_current_game()
|
||
if game is None or game["status"] != "placing":
|
||
return jsonify({"error": "Kein aktives Platzierungsspiel."}), 400
|
||
|
||
data = request.get_json(force=True)
|
||
ship_id = data.get("ship_id")
|
||
row = data.get("row")
|
||
col = data.get("col")
|
||
orientation = data.get("orientation", "h")
|
||
|
||
ship = FLEET_BY_ID.get(ship_id)
|
||
if ship is None:
|
||
return jsonify({"error": "Unbekannter Wappler."}), 400
|
||
|
||
board = game["player_board"]
|
||
already_placed = any(ship_id in row_ for row_ in board)
|
||
if already_placed:
|
||
return jsonify({"error": "Dieser Wappler steht schon auf dem Feld."}), 400
|
||
|
||
if not can_place(board, row, col, ship["size"], orientation):
|
||
return jsonify({"error": "Hier passt der Wappler nicht hin."}), 400
|
||
|
||
place_ship(board, ship_id, row, col, ship["size"], orientation)
|
||
game["player_board"] = board
|
||
|
||
placed_ids = {cell for row_ in board for cell in row_ if cell is not None}
|
||
if len(placed_ids) == len(FLEET):
|
||
game["message"] = "Alle Wappler stehen bereit! Klicke auf 'Kampf beginnen'."
|
||
else:
|
||
game["message"] = f"{ship['name']} platziert."
|
||
|
||
save_game(game)
|
||
return jsonify(build_state_payload(game))
|
||
|
||
|
||
@app.route("/api/random_place", methods=["POST"])
|
||
def api_random_place():
|
||
game = get_current_game()
|
||
if game is None or game["status"] != "placing":
|
||
return jsonify({"error": "Kein aktives Platzierungsspiel."}), 400
|
||
game["player_board"] = random_place_all_ships()
|
||
game["message"] = "Alle Wappler wurden zufällig aufgestellt."
|
||
save_game(game)
|
||
return jsonify(build_state_payload(game))
|
||
|
||
|
||
@app.route("/api/reset_placement", methods=["POST"])
|
||
def api_reset_placement():
|
||
game = get_current_game()
|
||
if game is None or game["status"] != "placing":
|
||
return jsonify({"error": "Kein aktives Platzierungsspiel."}), 400
|
||
game["player_board"] = empty_grid()
|
||
game["message"] = "Aufstellung zurückgesetzt."
|
||
save_game(game)
|
||
return jsonify(build_state_payload(game))
|
||
|
||
|
||
@app.route("/api/start", methods=["POST"])
|
||
def api_start():
|
||
game = get_current_game()
|
||
if game is None or game["status"] != "placing":
|
||
return jsonify({"error": "Kein aktives Platzierungsspiel."}), 400
|
||
|
||
placed_ids = {cell for row_ in game["player_board"] for cell in row_ if cell is not None}
|
||
if len(placed_ids) != len(FLEET):
|
||
return jsonify({"error": "Bitte alle Wappler platzieren, bevor der Kampf beginnt."}), 400
|
||
|
||
game["status"] = "playing"
|
||
game["message"] = "Der Kampf hat begonnen! Du bist am Zug – ziehe in die feindlichen Berge."
|
||
save_game(game)
|
||
return jsonify(build_state_payload(game))
|
||
|
||
|
||
@app.route("/api/fire", methods=["POST"])
|
||
def api_fire():
|
||
game = get_current_game()
|
||
if game is None or game["status"] != "playing":
|
||
return jsonify({"error": "Das Spiel läuft gerade nicht."}), 400
|
||
|
||
data = request.get_json(force=True)
|
||
row, col = data.get("row"), data.get("col")
|
||
if row is None or col is None or not (0 <= row < BOARD_SIZE and 0 <= col < BOARD_SIZE):
|
||
return jsonify({"error": "Ungültiges Feld."}), 400
|
||
|
||
if game["player_shots"][row][col] is not None:
|
||
return jsonify({"error": "Dorthin hast du schon geschossen."}), 400
|
||
|
||
board = game["computer_board"]
|
||
shots = game["player_shots"]
|
||
hit = board[row][col] is not None
|
||
shots[row][col] = "hit" if hit else "miss"
|
||
game["player_shots"] = shots
|
||
|
||
player_result = {"row": row, "col": col, "hit": hit, "sunk_ship": None}
|
||
if hit:
|
||
ship_id = board[row][col]
|
||
if ship_is_sunk(board, shots, ship_id):
|
||
player_result["sunk_ship"] = FLEET_BY_ID[ship_id]["name"]
|
||
|
||
computer_result = None
|
||
if all_ships_sunk(game["computer_board"], game["player_shots"]):
|
||
game["status"] = "player_win"
|
||
game["message"] = "Sieg! Du hast alle Wappler des Computers versenkt."
|
||
else:
|
||
computer_result = computer_take_shot(game)
|
||
if all_ships_sunk(game["player_board"], game["computer_shots"]):
|
||
game["status"] = "computer_win"
|
||
game["message"] = "Niederlage! Der Computer hat all deine Wappler besiegt."
|
||
else:
|
||
msgs = []
|
||
msgs.append(
|
||
f"Treffer bei {player_result['row']},{player_result['col']}!"
|
||
if player_result["hit"]
|
||
else "Daneben."
|
||
)
|
||
if player_result["sunk_ship"]:
|
||
msgs.append(f"{player_result['sunk_ship']} wurde versenkt!")
|
||
msgs.append(
|
||
"Der Computer trifft dich!" if computer_result["hit"] else "Der Computer schießt daneben."
|
||
)
|
||
if computer_result["sunk_ship"]:
|
||
msgs.append(f"Dein {computer_result['sunk_ship']} wurde versenkt!")
|
||
game["message"] = " ".join(msgs)
|
||
|
||
save_game(game)
|
||
payload = build_state_payload(game)
|
||
payload["player_shot"] = player_result
|
||
payload["computer_shot"] = computer_result
|
||
return jsonify(payload)
|
||
|
||
|
||
init_db()
|
||
|
||
if __name__ == "__main__":
|
||
app.run(host="0.0.0.0", debug=True, port=5000)
|