융합_로보테크 AI 자율주행 로봇 개발자 과정-26/01/26

지난주 프로젝트 이어서

 

DB x UI x

스레드 관리

세마포어

 

마피아게임

기본 8명

초과시 입장 불가

 

마피아2

경찰1

의사1

시민4(기자1 포함)

 

대기방: 클라이언트 => 서버

참여시 기본 메뉴얼 제공

스레드 생성

(if 인원 다 차면 30초 뒤에 시작한다고 서버에서 알림)

데이터베이스 훼손 안되게

(처음부터 닉네임 제공)

 

닉네임 입력(인원 8명 차면 이후부터 거절) ->

메뉴얼 ->

대기방(카운트) ->

직업 배정(랜덤) ->

반복(

    단체 채팅1 ->

    (밤전환)투표0.5(시간안에 못하면 패스) -> 

    변론(최다 투표자 말하기 + 죽일지 말지 Y/N) ->

    직업별 능력 사용0.5(기,마,경,의 순으로) ->

    투표 결과 발표0.5

) -> 

게임 종료

(if 마피아 전부 중도에 나가면 종료, 아니면 일단 진행 + 남은 인원의 절반 이상이 마피아면 마피아 승)

---

mafia

ㄴ server

    ㄴ main

    ㄴ model

    ㄴ view
    ㄴ controller

ㄴ client

 

Model: 게임 규칙과 상태

누가 참가했고(플레이어 목록), 누가 살아있고(생존자), 누가 무슨 직업인지(역할), 현재가 어떤 페이즈인지(낮/밤 등), 투표/능력 결과가 무엇인지 같은 도메인 상태를 보관하고 규칙대로 변경
네트워크, 소켓, 스레드, 출력 포맷 X

View: 플레이어에게 보여줄 텍스트

안내문, 시스템 공지, 채팅 브로드캐스트, 귓속말 같은 메시지 표현
실제 전송은 Sender 인터페이스(Controller)를 통해

Controller: 소켓 서버 + 스레드 + 입력 해석 + 게임 진행 담당
클라이언트 접속/퇴장, 닉네임 등록, 정원 8명 제한(세마포어), 수신 메시지 파싱(/w, /vote 등), 페이즈 타이머, 게임 루프를 돌리며 Model을 호출하고 View로 안내 뿌림

Main: 설정하고 서버 실행

 

Client: 서버와의 통신만 담당하는 단순 입출력
서버에 TCP로 연결, 사용자 입력을 서버로 전송, 서버로부터 수신한 메시지를 그대로 출력

---

main

import logging
from controller import MafiaServer, ServerConfig


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )

    cfg = ServerConfig(host="0.0.0.0", port=50007)
    MafiaServer(cfg).serve_forever()


if __name__ == "__main__":
    main()

model

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Optional, Set
import random

class Role():
    pass

class Phase():
    pass

class Player():
    pass

class GameRuleError():
    pass

class GameState():
    # 상태/프로퍼티: capacity, phase, player_count, alive_players,
    # 참가/퇴장: join, leave,
    # 역할/페이즈: assign_roles_default_8, role_of, is_alive, set_phase,
    # 낮투표: sub_day_vote, tally_day_vote, clr_day_vote,
    # 직업 능력: mafia_target, doctor_save, police_check,
    # 밤 처리: resolve_night,
    # 내부: pick_mafia_maj_target, pick_any_save_target, majority, tally vote, kill, require_alive,
    # 승리/종료: mafia_all_left, winner, check_end_conditions
    pass

view

from __future__ import annotations

from dataclasses import dataclass
from typing import Dict, Optional, Protocol
import socket

class Sender(Protocol):
    # send_to, broadcast
    pass

@dataclass()
class Manual:
    pass

class TextView:
    # init, show_manual, info,
    # sys_broadcast, chat_broadcast,
    # whisper, phase, roster
    pass

controller

from __future__ import annotations

import socket
import threading
import time
import logging
from dataclasses import dataclass
from typing import Dict, Optional, Tuple

from model import GameRuleError, GameState, Phase, Role
from view import TextView

@dataclass()
class ServerConfig:
    host: str
    port: int

class LineProtocol:
    # init, encode, feed, pop_line
    pass

class MafiaServer:
    # init, server_forever, send_to, broadcast, handle_client,
    # if_start_game, run_game_loop,
    # night_sequence, day_sequence,
    # route_message, conn_of, recv_line
    pass

 

MVC 적용하기엔 시간 부족 + TCP에는 안 맞음

---

client에 exit만 추가

import socket

HOST = '~~~~~'
PORT = 50007

print("안내메시지(메뉴얼~~~~~)")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_socket:
    client_socket.connect((HOST, PORT))

    while True:
        message = input("Client: ").strip()
        client_socket.sendall(message.encode("utf-8"))

        data = client_socket.recv(1024)
        reply = data.decode("utf-8")
        print("Server:", reply)

        if message == "exit":
            break

---

설계

 

1. 초기설정 및 클라이언트 접속

• 서버는 클라이언트 접속 시 정원 초과 여부를 먼저 검사합니다.
• 정원 이내일 경우 딕셔너리를 통해 정보를 저장합니다.
• 딕셔너리 예시: {직업 : IP주소}, {닉네임 : IP주소}
• 닉네임은 서버에서 처음부터 랜덤하게 부여합니다 (예: 토끼 : 192.0.0.1).
• 마지막 클라이언트가 접속하여 정원이 차면, 5초 뒤에 게임을 시작한다는 전체 공지를 보냅니다.

2. 시스템 모니터링 및 에러 처리

• 별도의 비동기 루프(async while True)를 통해 클라이언트 수와 이탈자 발생 여부를 지속적으로 체크합니다.
• 게임이 시작되면 해당 루프를 본격적으로 돌려 게임 상태를 동기화합니다.

 

3. 게임 메인 루프: 낮 단계

• 단체 채팅 (40초): 클라이언트가 채팅을 보내면 서버가 읽어와서 IP주소와 매칭된 닉네임을 붙여 모든 접속자에게 전송(sendall)합니다.
• 서버는 시간을 측정하여 30초가 지나면 클라이언트에게 투표 시간이 되었다는 메시지를 전체 공지합니다.

 

4. 게임 메인 루프: 투표 단계

• 투표 기능 (30초): 서버가 현재 생존한 닉네임 리스트를 먼저 전체 공지합니다.
• 클라이언트가 닉네임을 입력하여 전송하면 서버는 해당 닉네임의 득표수를 계산합니다.
• 각 클라이언트에게 토큰을 부여하여 1인 1표만 가능하도록 중복 투표를 방지합니다.
• 서버는 시간을 측정하여 10초가 지나면 투표 마감 메시지를 보냅니다.
• 최다 득표자가 여러 명일 경우 재투표를 실시합니다.
• 투표 결과에 따라 탈락자가 발생하며, 이때 종료 조건을 검사합니다.

 

5. 게임 메인 루프: 밤 단계 (직업별 능력 사용 30초)

• 시민: 별도 행동 없이 대기합니다.
• 기자: 2번째 밤부터 가능하며, 닉네임 리스트를 받아 대상을 지목하면 정보를 수집합니다.
• 마피아: 닉네임 리스트를 받고 전용 채팅을 통해 대상을 지목합니다.
• 경찰: 닉네임 리스트를 받아 대상을 지목하면 마피아 여부를 확인합니다.
• 의사: 닉네임 리스트를 받아 마피아로부터 살릴 대상을 지목합니다.

 

6. 직업별 결과 판정 및 로직

• 마피아가 지목한 사람과 의사가 지목한 사람이 같으면 "의사가 누구를 살렸습니다"를 결과에 저장합니다.
• 지목한 사람이 서로 다르면 해당 클라이언트를 사망(alive = false) 처리합니다.
• 기자가 선택한 대상의 직업 정보나 경찰이 선택한 대상의 마피아 여부 정보를 각각 저장합니다.
• 밤 단계 종료 후 다시 종료 조건을 검사합니다.

 

7. 게임 종료 조건 및 마무리

• 마피아가 모두 죽거나, 시민 측 인원이 마피아 수와 같아지거나 다 죽으면 게임을 종료합니다.
• 게임 종료 시 클라이언트에게 안내 메시지를 먼저 보냅니다.
• 클라이언트 소켓들을 먼저 정리하여 연결을 끊습니다.
• 모든 클라이언트 소켓 정리가 완료되면 최종적으로 서버 소켓을 닫고 프로세스를 종료합니다.

---

from __future__ import annotations

import asyncio
import random
import time
from collections import Counter
from dataclasses import dataclass, field
from typing import Dict, Optional, Set, Tuple, List

ANIMALS = ["토끼", "고양이", "사자", "강아지", "악어", "코끼리", "호랑이", "개미"]
JOBS = (["마피아"] * 2 + ["경찰"] * 1 + ["의사"] * 1 + ["기자"] * 1 + ["시민"] * 3)

MANUAL = (
    "메뉴얼:\n"
    "  /help                도움말\n"
    "  /vote 닉             (낮 투표)\n"
    "  /kill 닉             (마피아) 살인 대상(밤)\n"
    "  /heal 닉             (의사) 살릴 대상(밤)\n"
    "  /check 닉            (경찰) 조사(밤) -> 즉시 본인에게만 직업 공개\n"
    "  /peek 닉             (기자) 2번째 밤부터 조사(다음날 아침 전체 공개)\n"
    "  /m 메시지            (마피아) 밤 전용 채팅\n"
    "  exit                 종료\n"
    "\n"
)

@dataclass(frozen=True)
class Config:
    host: str = "0.0.0.0"
    port: int = 50007
    capacity: int = 8

    start_delay: int = 5

    day_chat: int = 40
    day_chat_warn_after: int = 30  # 30초 경과(=남은 10초) 투표 예고

    day_vote: int = 30
    vote_warn_remaining: int = 10
    revote: int = 20

    night: int = 30
    monitor_interval: float = 0.5

@dataclass
class State:
    # 보호 락(공유자원 보호)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    capacity: int = 8

    # 서버 종료 제어
    shutdown: bool = False

    # 게임 상태
    started: bool = False
    phase: str = "lobby"  # lobby/day_chat/day_vote/night/ended
    night_count: int = 0

    # 세션
    writer_by_nick: Dict[str, asyncio.StreamWriter] = field(default_factory=dict)
    nick_by_writer: Dict[asyncio.StreamWriter, str] = field(default_factory=dict)

    # 설계 요구 딕셔너리 구조
    ip_by_nick: Dict[str, str] = field(default_factory=dict)          # {닉: IP}
    ips_by_job: Dict[str, Set[str]] = field(default_factory=dict)     # {직업: {IP...}}
    job_by_nick: Dict[str, str] = field(default_factory=dict)         # {닉: 직업}
    alive: Set[str] = field(default_factory=set)

    # 낮 투표
    vote_token: Dict[str, bool] = field(default_factory=dict)
    day_votes: Dict[str, str] = field(default_factory=dict)

    # 밤 행동
    night_kill_votes: Dict[str, str] = field(default_factory=dict)  # {mafia_nick: target} -> 다수결
    night_heal: Dict[str, str] = field(default_factory=dict)        # {doctor_nick: target}

    # 기자: 밤 예약 -> 다음날 아침 전체 공개 (기자 생사 무관)
    reporter_peek_target: Optional[str] = None
    pending_report_reveal: Optional[str] = None


class MafiaServer:
    def __init__(self, cfg: Config):
        self.cfg = cfg
        self.s = State(capacity=cfg.capacity)
        self.rng = random.Random()

        # 닉/직업 풀(랜덤 부여)
        self.nick_pool = ANIMALS[:]
        self.job_pool = JOBS[:]
        self.rng.shuffle(self.nick_pool)
        self.rng.shuffle(self.job_pool)

        self.engine_task: Optional[asyncio.Task] = None
        self.monitor_task: Optional[asyncio.Task] = None
        self.server: Optional[asyncio.base_events.Server] = None

    # ----------------- I/O helpers -----------------
    async def send_writer(self, w: asyncio.StreamWriter, text: str) -> None:
        try:
            w.write(text.encode())
            await w.drain()
        except Exception:
            pass

    async def send_to(self, nick: str, text: str) -> None:
        async with self.s.lock:
            w = self.s.writer_by_nick.get(nick)
        if w:
            await self.send_writer(w, text)

    async def broadcast(self, text: str) -> None:
        async with self.s.lock:
            writers = list(self.s.nick_by_writer.keys())
        for w in writers:
            await self.send_writer(w, text)

    async def mafia_broadcast(self, text: str) -> None:
        async with self.s.lock:
            mafia_nicks = [n for n in self.s.alive if self.s.job_by_nick.get(n) == "마피아"]
            writers = [self.s.writer_by_nick.get(n) for n in mafia_nicks]
        for w in writers:
            if w:
                await self.send_writer(w, text)

    # ----------------- lifecycle -----------------
    async def shutdown(self) -> None:
        async with self.s.lock:
            if self.s.shutdown:
                return
            self.s.shutdown = True

            writers = list(self.s.nick_by_writer.keys())

        # 클라 연결 닫기
        for w in writers:
            try:
                w.close()
            except Exception:
                pass

        # 서버 닫기(accept loop 종료)
        if self.server:
            self.server.close()
            try:
                await self.server.wait_closed()
            except Exception:
                pass

    async def start(self) -> None:
        self.server = await asyncio.start_server(self.handle_client, self.cfg.host, self.cfg.port)
        print(f"서버 실행: {self.cfg.host}:{self.cfg.port}")

        # 모니터링 코루틴(설계 요구)
        self.monitor_task = asyncio.create_task(self.monitor_loop())

        async with self.server:
            await self.server.serve_forever()

    # ----------------- monitor (async while True) -----------------
    async def monitor_loop(self) -> None:
        while True:
            await asyncio.sleep(self.cfg.monitor_interval)

            async with self.s.lock:
                if self.s.shutdown:
                    return
                count = len(self.s.writer_by_nick)
                started = self.s.started
                phase = self.s.phase

            # lobby에서 8명 충족 & 미시작이면 엔진 시작
            if (not started) and phase == "lobby" and count >= self.cfg.capacity:
                if self.engine_task is None or self.engine_task.done():
                    self.engine_task = asyncio.create_task(self.game_engine())

    # ----------------- registration / leave -----------------
    async def register(self, w: asyncio.StreamWriter, peer_ip: str) -> Optional[str]:
        async with self.s.lock:
            if len(self.s.writer_by_nick) >= self.cfg.capacity or not self.nick_pool:
                return None

            nick = self.nick_pool.pop()
            job = self.job_pool.pop()

            self.s.writer_by_nick[nick] = w
            self.s.nick_by_writer[w] = nick

            self.s.ip_by_nick[nick] = peer_ip
            self.s.job_by_nick[nick] = job
            self.s.alive.add(nick)
            self.s.ips_by_job.setdefault(job, set()).add(peer_ip)

            return nick

    async def leave_as_dead(self, w: asyncio.StreamWriter) -> None:
        async with self.s.lock:
            nick = self.s.nick_by_writer.pop(w, None)
            if not nick:
                return

            # 이탈자 즉시 사망 처리(승리 판정 안정화)
            self.s.alive.discard(nick)

            job = self.s.job_by_nick.get(nick)
            ip = self.s.ip_by_nick.get(nick)

            self.s.writer_by_nick.pop(nick, None)
            # job/ip는 기록 유지 가능하나, ips_by_job은 정리
            if job and ip and job in self.s.ips_by_job:
                self.s.ips_by_job[job].discard(ip)

        await self.broadcast(f"[퇴장] {nick} (이탈로 사망 처리)\n")

    # ----------------- rules helpers -----------------
    async def my_job_phase_alive(self, nick: str) -> Tuple[str, str, bool, int]:
        async with self.s.lock:
            return (
                self.s.job_by_nick.get(nick, ""),
                self.s.phase,
                (nick in self.s.alive),
                self.s.night_count,
            )

    async def alive_list(self) -> List[str]:
        async with self.s.lock:
            return sorted(self.s.alive)

    def pick_majority_or_random(self, choices: List[str]) -> Optional[str]:
        if not choices:
            return None
        c = Counter(choices)
        best = c.most_common(1)[0][1]
        tied = [k for k, v in c.items() if v == best]
        return tied[0] if len(tied) == 1 else self.rng.choice(tied)

    async def check_winner(self) -> Optional[str]:
        async with self.s.lock:
            alive = set(self.s.alive)
            mafia = {n for n in alive if self.s.job_by_nick.get(n) == "마피아"}
            citizens = alive - mafia

            if self.s.job_by_nick and not mafia:
                return "시민"
            if mafia and len(mafia) >= len(citizens):
                return "마피아"
            return None

    # ----------------- game engine -----------------
    async def game_engine(self) -> None:
        await self.broadcast(f"[시스템] 정원 충족. {self.cfg.start_delay}초 뒤 게임 시작.\n")
        await asyncio.sleep(self.cfg.start_delay)

        async with self.s.lock:
            # 시작 직전 인원 부족이면 취소
            if len(self.s.writer_by_nick) < self.cfg.capacity:
                await self.broadcast("[시스템] 인원 부족으로 시작 취소. 다시 8명 모이면 시작합니다.\n")
                return

            self.s.started = True
            self.s.phase = "day_chat"

        await self.broadcast("[시스템] 게임 시작!\n")

        # 개인 직업 안내
        async with self.s.lock:
            items = list(self.s.job_by_nick.items())
        for nick, job in items:
            await self.send_to(nick, f"[개인] 당신의 직업은 {job} 입니다.\n")

        while True:
            winner = await self.check_winner()
            if winner:
                async with self.s.lock:
                    self.s.phase = "ended"
                await self.broadcast(f"[게임 종료] 승리: {winner}\n")
                await self.shutdown()
                return

            # 아침: 기자 공개 예약 처리
            async with self.s.lock:
                msg = self.s.pending_report_reveal
                self.s.pending_report_reveal = None
            if msg:
                await self.broadcast(msg)

            await self.phase_day_chat()
            if await self.check_winner():
                continue

            eliminated = await self.phase_day_vote()
            await self.broadcast(f"[결과] 낮 투표로 {eliminated} 탈락\n" if eliminated else "[결과] 낮 투표 패스(처형 없음)\n")
            if await self.check_winner():
                continue

            killed, saved = await self.phase_night()
            if saved:
                await self.broadcast(f"[결과] 의사가 {saved}를 살렸습니다.\n")
            if killed:
                await self.broadcast(f"[결과] 밤에 {killed} 사망\n")
            if not killed and not saved:
                await self.broadcast("[결과] 밤 사망자 없음\n")

    async def phase_day_chat(self) -> None:
        async with self.s.lock:
            self.s.phase = "day_chat"
        await self.broadcast(f"[페이즈] 낮 채팅 시작 ({self.cfg.day_chat}초)\n")

        start = time.time()
        warned = False
        while time.time() - start < self.cfg.day_chat:
            async with self.s.lock:
                if self.s.shutdown:
                    return
            if not warned and (time.time() - start) >= self.cfg.day_chat_warn_after:
                await self.broadcast("[시스템] 10초 뒤 투표 시간입니다.\n")
                warned = True
            await asyncio.sleep(0.2)

    async def phase_day_vote(self) -> Optional[str]:
        async with self.s.lock:
            self.s.phase = "day_vote"
            self.s.day_votes.clear()
            self.s.vote_token = {n: True for n in self.s.alive}
            alive_str = ", ".join(sorted(self.s.alive))

        await self.broadcast(f"[페이즈] 낮 투표 시작 ({self.cfg.day_vote}초) /vote 닉\n")
        await self.broadcast(f"[생존자] {alive_str}\n")

        # 마감 10초 전 공지 포함 타이머
        start = time.time()
        warned = False
        while time.time() - start < self.cfg.day_vote:
            async with self.s.lock:
                if self.s.shutdown:
                    return None
            remaining = self.cfg.day_vote - (time.time() - start)
            if not warned and remaining <= self.cfg.vote_warn_remaining:
                await self.broadcast("[시스템] 투표 마감 10초 전입니다.\n")
                warned = True
            await asyncio.sleep(0.2)

        tied = await self.tally_top()
        if not tied:
            return None
        if len(tied) == 1:
            target = tied[0]
            async with self.s.lock:
                self.s.alive.discard(target)
            return target

        # 재투표 1회
        await self.broadcast(f"[시스템] 동점 발생: {', '.join(tied)} (재투표 1회)\n")
        async with self.s.lock:
            self.s.day_votes.clear()
            self.s.vote_token = {n: True for n in self.s.alive}

        start2 = time.time()
        warned2 = False
        while time.time() - start2 < self.cfg.revote:
            async with self.s.lock:
                if self.s.shutdown:
                    return None
            remaining = self.cfg.revote - (time.time() - start2)
            if not warned2 and remaining <= self.cfg.vote_warn_remaining:
                await self.broadcast("[시스템] 재투표 마감 10초 전입니다.\n")
                warned2 = True
            await asyncio.sleep(0.2)

        tied2 = await self.tally_top(allow_only=set(tied))
        if not tied2:
            return None
        target = tied2[0] if len(tied2) == 1 else self.rng.choice(tied2)
        if len(tied2) > 1:
            await self.broadcast(f"[시스템] 재투표도 동점 -> 랜덤 처형: {target}\n")

        async with self.s.lock:
            self.s.alive.discard(target)
        return target

    async def tally_top(self, allow_only: Optional[Set[str]] = None) -> List[str]:
        async with self.s.lock:
            counts = Counter()
            for voter, target in self.s.day_votes.items():
                if voter not in self.s.alive:
                    continue
                if target not in self.s.alive:
                    continue
                if allow_only is not None and target not in allow_only:
                    continue
                counts[target] += 1

        if not counts:
            return []
        best = counts.most_common(1)[0][1]
        return [n for n, c in counts.items() if c == best]

    async def phase_night(self) -> Tuple[Optional[str], Optional[str]]:
        async with self.s.lock:
            self.s.phase = "night"
            self.s.night_count += 1
            night_no = self.s.night_count

            self.s.night_kill_votes.clear()
            self.s.night_heal.clear()
            self.s.reporter_peek_target = None

            alive_str = ", ".join(sorted(self.s.alive))

        await self.broadcast(f"[페이즈] 밤 {night_no} 시작 ({self.cfg.night}초)\n")
        await self.broadcast(f"[생존자] {alive_str}\n")
        await self.broadcast("[시스템] 마피아:/kill, 의사:/heal, 경찰:/check, 기자(2번째 밤부터):/peek, 마피아채팅:/m\n")

        await asyncio.sleep(self.cfg.night)

        # 밤 판정(마피아 다수결, 의사 다수결)
        async with self.s.lock:
            kill_target = self.pick_majority_or_random(list(self.s.night_kill_votes.values()))
            heal_target = self.pick_majority_or_random(list(self.s.night_heal.values()))

            killed: Optional[str] = None
            saved: Optional[str] = None

            if kill_target:
                if heal_target and heal_target == kill_target:
                    saved = heal_target
                else:
                    self.s.alive.discard(kill_target)
                    killed = kill_target

            # 기자 공개 예약(기자 생사 무관)
            if self.s.reporter_peek_target:
                t = self.s.reporter_peek_target
                role = self.s.job_by_nick.get(t, "알수없음")
                self.s.pending_report_reveal = f"[특보] 기자 조사 결과: {t}의 직업은 {role} 입니다.\n"
                self.s.reporter_peek_target = None

        return killed, saved

    # ----------------- client handler -----------------
    async def handle_client(self, r: asyncio.StreamReader, w: asyncio.StreamWriter) -> None:
        peer = w.get_extra_info("peername")
        peer_ip = peer[0] if isinstance(peer, tuple) and peer else "?"

        nick = await self.register(w, peer_ip)
        if nick is None:
            await self.send_writer(w, "정원(8명) 초과\n")
            try:
                w.close()
            except Exception:
                pass
            return

        # 접속 즉시 안내(push)
        await self.send_to(nick, MANUAL)
        await self.send_to(nick, f"[시스템] 당신의 닉네임: {nick} / IP: {peer_ip}\n")
        await self.broadcast(f"[입장] {nick} ({peer_ip})\n")

        async with self.s.lock:
            if not self.s.started and self.s.phase == "lobby":
                await self.broadcast(f"[대기실] 현재 인원: {len(self.s.writer_by_nick)}/{self.cfg.capacity}\n")

        try:
            while True:
                line = await r.readline()
                if not line:
                    return
                msg = line.decode(errors="replace").strip()
                if not msg:
                    continue

                if msg == "exit":
                    await self.send_to(nick, "bye\n")
                    return

                if msg == "/help":
                    await self.send_to(nick, MANUAL)
                    continue

                job, phase, alive, night_no = await self.my_job_phase_alive(nick)

                # 죽은 사람 제한 강화
                if not alive:
                    await self.send_to(nick, "당신은 사망했습니다. 관전만 가능합니다.\n")
                    continue

                # 낮 채팅
                if phase == "day_chat" and not msg.startswith("/"):
                    async with self.s.lock:
                        ip = self.s.ip_by_nick.get(nick, "?")
                    await self.broadcast(f"{nick}({ip}): {msg}\n")
                    continue

                # 낮 투표
                if msg.startswith("/vote "):
                    if phase != "day_vote":
                        await self.send_to(nick, "지금은 투표 시간이 아닙니다.\n")
                        continue
                    target = msg.replace("/vote", "", 1).strip()
                    async with self.s.lock:
                        if target not in self.s.alive:
                            await self.send_to(nick, "대상이 생존자가 아닙니다.\n")
                            continue
                        if not self.s.vote_token.get(nick, False):
                            await self.send_to(nick, "이미 투표했습니다.\n")
                            continue
                        self.s.day_votes[nick] = target
                        self.s.vote_token[nick] = False
                    await self.send_to(nick, f"투표 완료: {target}\n")
                    continue

                # night 아닌데 밤 명령이면 안내
                if phase != "night":
                    await self.send_to(nick, "명령을 확인하세요. /help\n")
                    continue

                # 밤: 시민 제한
                if job == "시민":
                    await self.send_to(nick, "시민은 밤에 행동이 없습니다.\n")
                    continue

                # 마피아 채팅
                if msg.startswith("/m "):
                    if job != "마피아":
                        await self.send_to(nick, "마피아만 사용할 수 있습니다.\n")
                        continue
                    text = msg.replace("/m", "", 1).strip()
                    if text:
                        await self.mafia_broadcast(f"[마피아] {nick}: {text}\n")
                    continue

                # 마피아 kill(다수결 투표)
                if msg.startswith("/kill "):
                    if job != "마피아":
                        await self.send_to(nick, "마피아만 가능합니다.\n")
                        continue
                    target = msg.replace("/kill", "", 1).strip()
                    async with self.s.lock:
                        if target not in self.s.alive:
                            await self.send_to(nick, "대상이 생존자가 아닙니다.\n")
                            continue
                        self.s.night_kill_votes[nick] = target
                    await self.send_to(nick, f"살인 투표 등록: {target}\n")
                    continue

                # 의사 heal
                if msg.startswith("/heal "):
                    if job != "의사":
                        await self.send_to(nick, "의사만 가능합니다.\n")
                        continue
                    target = msg.replace("/heal", "", 1).strip()
                    async with self.s.lock:
                        if target not in self.s.alive:
                            await self.send_to(nick, "대상이 생존자가 아닙니다.\n")
                            continue
                        self.s.night_heal[nick] = target
                    await self.send_to(nick, f"치료 대상 지정: {target}\n")
                    continue

                # 경찰 check: 즉시 본인에게만 직업 공개
                if msg.startswith("/check "):
                    if job != "경찰":
                        await self.send_to(nick, "경찰만 가능합니다.\n")
                        continue
                    target = msg.replace("/check", "", 1).strip()
                    async with self.s.lock:
                        if target not in self.s.alive:
                            await self.send_to(nick, "대상이 생존자가 아닙니다.\n")
                            continue
                        role = self.s.job_by_nick.get(target, "알수없음")
                    await self.send_to(nick, f"[개인] 조사 결과: {target}의 직업은 {role} 입니다.\n")
                    continue

                # 기자 peek: 2번째 밤부터 가능, 다음날 아침 전체공개 예약
                if msg.startswith("/peek "):
                    if job != "기자":
                        await self.send_to(nick, "기자만 가능합니다.\n")
                        continue
                    if night_no < 2:
                        await self.send_to(nick, "기자는 2번째 밤부터 조사 가능합니다.\n")
                        continue
                    target = msg.replace("/peek", "", 1).strip()
                    async with self.s.lock:
                        if target not in self.s.alive:
                            await self.send_to(nick, "대상이 생존자가 아닙니다.\n")
                            continue
                        self.s.reporter_peek_target = target
                    await self.send_to(nick, f"기자 조사 예약 완료: {target} (다음날 아침 전체 공개)\n")
                    continue

                await self.send_to(nick, "명령을 확인하세요. /help\n")

        except Exception:
            return
        finally:
            await self.leave_as_dead(w)

def main():
    cfg = Config()
    srv = MafiaServer(cfg)
    asyncio.run(srv.start())

if __name__ == "__main__":
    main()

from __future__ import annotations

import socket
import threading


HOST = "192.168.0.204"
PORT = 50007
ENCODING = "utf-8"


def recv_loop(sock: socket.socket, stop: threading.Event) -> None:
    try:
        while not stop.is_set():
            data = sock.recv(4096)
            if not data:
                print("Server: (연결 종료)")
                stop.set()
                return
            msg = data.decode(ENCODING, errors="replace")
            print(msg, end="" if msg.endswith("\n") else "\n")
    except OSError:
        stop.set()


def main() -> None:
    stop = threading.Event()

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.connect((HOST, PORT))
            print("종료: exit")

            t = threading.Thread(target=recv_loop, args=(s, stop), daemon=True)
            t.start()

            while not stop.is_set():
                try:
                    message = input("Client: ").strip()
                except (EOFError, KeyboardInterrupt):
                    message = "exit"

                if not message:
                    continue

                try:
                    s.sendall(message.encode(ENCODING))
                except OSError:
                    print("Server: (전송 실패/연결 종료)")
                    stop.set()
                    break

                if message == "exit":
                    stop.set()
                    break

    except ConnectionRefusedError:
        print("Server: 접속 실패(서버가 꺼져있거나 주소/포트가 틀림)")
    except OSError as e:
        print(f"Server: 네트워크 오류({e})")


if __name__ == "__main__":
    main()