[MistServer] 스트림 관리 API 실습 (Python)

반응형

MistServer 스트림 관리 API 실습 (Python)

"본 글은 과거 티스토리에서 발행했던 콘텐츠를 기반으로, 최신 정보를 반영해 새롭게 정리한 업데이트 버전입니다."

 

📚 목차
1. 인증과 준비 - API 인증 방식, 요청 포맷, 공통 유틸
2. 등록과 업데이트 - addstream
3. 제어와 정리 - 삭제/강제종료/자동정리/세션종료
4. 조회와 모니터링 - active_streams, stats_streams
✔ 마무리

 

MistStream 스트림 API 삽화 이미지

1. 인증과 준비

▸ API2 스타일은 하나의 JSON에 "authorize": {...} 블록과 원하는 동작을 함께 보냅니다.

▸ 비밀번호는 challenge 기반 이중 MD5를 사용합니다:
password = MD5( MD5(원문비번) + challenge )
(challenge는 서버가 제공. 실제 수신·사용 방식은 환경에 따라 다를 수 있으며, 아래 함수는 해시 계산만 담당합니다.)

 

✔️ 공통 사용 함수

import hashlib
import requests

# ==============================
# 🔹 설정 상수
# ==============================
SERVER_URL = "http://101.1.xxx.xxx:4242/api2"  # MistServer API2 엔드포인트
USERNAME = "ceri"                           # 로그인 사용자명
PASSWORD = "cericube"                       # 원문 비밀번호
HEADERS = {"content-type": "application/json"}  # HTTP 요청 헤더


# ==============================
# 🔹 인증 해시 생성 함수
# ==============================
def make_api2_password(plain_password: str, challenge: str) -> str:
    """
    MistServer API2 인증용 비밀번호 해시를 생성합니다.
    형식: MD5( MD5(원문 비밀번호) + challenge )

    Args:
        plain_password (str): 원문 비밀번호
        challenge (str): 서버에서 제공한 challenge 문자열
    Returns:
        str: 최종 해시된 비밀번호
    """
    # 1차 해시: 원문 비밀번호 → MD5 해시 문자열
    first_hash = hashlib.md5(plain_password.encode()).hexdigest()
    # 2차 해시: (1차 해시 + challenge) → MD5 해시 문자열
    final_hash = hashlib.md5((first_hash + challenge).encode()).hexdigest()
    return final_hash


# ==============================
# 🔹 API 요청 페이로드 생성
# ==============================
def merge_payload(passwd: str, payload: dict) -> dict:
    """
    API 요청에 필요한 authorize 블록과
    추가 동작 payload를 합칩니다.
    """
    return {
        "authorize": {
            "username": USERNAME,
            "password": passwd
        },
        **payload
    }


# ==============================
# 🔹 API 호출 함수
# ==============================
def request_api(payload: dict) -> dict:
    """
    주어진 payload로 API2 요청을 전송하고,
    응답을 JSON 형태로 반환합니다.
    """
    response = requests.post(SERVER_URL, json=payload, headers=HEADERS)
    response.raise_for_status()  # HTTP 오류 발생 시 예외 발생
    return response.json()


# ==============================
# 🔹 로그인 절차
# ==============================
def login() -> str:
    """
    1. 서버에서 challenge 값을 요청
    2. challenge 기반 인증 해시 생성
    3. 생성된 해시 반환
    """
    # 1️. 챌린지 요청
    initial_payload = merge_payload(PASSWORD, {})
    print(f"1. 챌린지 요청 페이로드: {initial_payload}")

    res = request_api(initial_payload)
    challenge = res.get("authorize", {}).get("challenge")
    print(f"1. Challenge 값 수신: {challenge}")

    # 2️. 인증 해시 생성
    auth_hash = make_api2_password(PASSWORD, challenge)
    print(f"2. 생성된 인증 해시: {auth_hash}")

    return auth_hash

 

2. 등록과 업데이트 - addstream

MistServer 3.8의 addstream API 함수는 스트림을 개별적으로 추가하거나 업데이트할 때 사용하는 함수입니다.

기존에 설정된 스트림들은 그대로 유지되고, 새로운 스트림만 추가되거나 기존 스트림이 업데이트됩니다.

빈 addstream 요청을 보내더라도 모든 스트림이 삭제되지 않으며, 응답으로는 전체 스트림 목록 대신 추가되거나 업데이트된 스트림만 반환됩니다.

이를 구분하기 위해 응답에 "incomplete list":1이 포함됩니다.

 

✔️ 코드 예시

# ==============================
# 🔹 addstream 실습 함수
# ==============================
def test_addstream(auth_hash: str) -> dict:
    """
    addstream 실습용 함수.
    - stream01/stream02 두 개의 push 입력을 등록(또는 업데이트)합니다.
    - 성공 시 응답 JSON을 그대로 반환합니다.

    Args:
        auth_hash (str): MD5(MD5(비번)+challenge)로 만든 최종 인증 해시

    Returns:
        dict: /api2 응답(JSON)
    """
    addstream_body = {
        "addstream": {
            "stream01": {
                "source": "push://",   # OBS 등 외부 송출을 받을 push 입력
                "stop_sessions": False, # True면 해당 스트림 관련 세션 즉시 종료
                "resume": 1,            # 중단 후 재개 허용
                "segmentsize": "1900"   # ms 단위 세그먼트 (예: 6000 => 6초)
            },
            "stream02": {
                "source": "push://",
                "stop_sessions": False,
                "resume": 1,
                "segmentsize": "1900"
            }
        }
    }

    # authorize 블록 병합
    payload = merge_payload(auth_hash, addstream_body)
    print(f"[addstream] 요청 페이로드: {payload}")

    # 요청 전송
    res = request_api(payload)
    print(f"[addstream] 응답: {res}")

    # ✅ 기대 포인트:
    # res["streams"] 내부에 "incomplete list": 1 플래그와 함께,
    # 방금 추가/변경된 stream01/stream02 정보가 포함될 수 있습니다.
    # (addstream은 전체 목록이 아니라 '변경된 항목만' 응답에 실리는 것이 정상)
    return res

 

✔️ 응답 예시

{
  "LTS": 1,
  "authorize": {
    "status": "OK"
  },
  "streams": {
    "incomplete list": 1,
    "stream01": {....},
    "stream02": {....}
  }
}

 

📌 스트림 전체 설정 : streams
스트림 목록을 전체 덮어쓰기(Overwrite) 방식으로 설정합니다.
한 번의 요청으로 모든 스트림을 교체하며, 빈 요청 시 모든 스트림이 삭제되므로 주의가 필요합니다.
대신, 개별 추가/삭제는 addstream, deletestream 사용 권장

반응형

 

3. 제어와 정리 — 삭제·강제종료·자동정리·세션종료

🔷 선택 삭제 - deletestream

지정한 스트림만 선택적으로 삭제하며, 다른 스트림에는 영향을 주지 않습니다

def test_deletestream(auth_hash: str) -> dict:
    """
    delete_stream 실습용 함수.
    - stream01 스트림을 삭제합니다.
    - 성공 시 응답 JSON을 그대로 반환합니다.

    Args:
        auth_hash (str): MD5(MD5(비번)+challenge)로 만든 최종 인증 해시

    Returns:
        dict: /api2 응답(JSON)
    """
    delete_stream_body = {
       "deletestream": ["stream03", "stream01"]
    }

    # authorize 블록 병합
    payload = merge_payload(auth_hash, delete_stream_body)
    print(f"[deletestream] 요청 페이로드: {payload}")

    # 요청 전송
    res = request_api(payload)
    print(f"[deletestream] 응답: {res}")

    return res

 

🔷 소스 포함 삭제 - deletestreamsource

스트림을 삭제하면서 연결된 소스 파일도 함께 삭제를 시도합니다.
HLS처럼 다중 파일(플레이리스트) 구조인 경우에는 모든 파일이 삭제되지 않을 수 있습니다.
DTSH 헤더 파일이 존재하면 해당 파일도 함께 삭제를 시도합니다.

def test_deletestreamsource(auth_hash: str) -> dict:
    """
    delete_stream_source 실습용 함수.
    - stream01 스트림의 소스를 삭제합니다.
    - 성공 시 응답 JSON을 그대로 반환합니다.

    Args:
        auth_hash (str): MD5(MD5(비번)+challenge)로 만든 최종 인증 해시

    Returns:
        dict: /api2 응답(JSON)
    """
    delete_stream_source_body = {
		"deletestreamsource": ["stream01"]
    }

    # authorize 블록 병합
    payload = merge_payload(auth_hash, delete_stream_source_body)
    print(f"[deletestreamsource] 요청 페이로드: {payload}")

    # 요청 전송
    res = request_api(payload)
    print(f"[deletestreamsource] 응답: {res}")

    return res

{
   "LTS":1,
   "authorize":{
      "status":"OK"
   },
   "deletestreamsource":{
      "stream01":"-1: Stream deleted, source remains"  # 응답코드
   }
}

응답코드:
0: No action taken
1: Source file deleted
2: Source file and dtsh deleted
  : Stream deleted, source remains
  : Stream and source file deleted
  : Stream, source file and dtsh deleted

음수 값: 서버 설정에서 스트림이 제거됨을 의미
양수 값: 스트림이 서버 설정에서 제거되지 않았음을 의미

 

🔷 완전 강제 종료 - nuke_stream

실행 중인 스트림을 정상 종료 → 강제 종료 → 메모리 정리의 3단계를 거쳐 완전히 종료합니다.
저작권 침해나 장애 발생 등 즉시 차단이 필요한 상황에서 효과적으로 사용할 수 있습니다.

def test_nuke_stream(auth_hash: str) -> dict:  
    """
    nuke_stream 실습용 함수.
    - stream01 스트림을 완전히 삭제합니다.
    - 성공 시 응답 JSON을 그대로 반환합니다.

    Args:
        auth_hash (str): MD5(MD5(비번)+challenge)로 만든 최종 인증 해시

    Returns:
        dict: /api2 응답(JSON)
    """
    nuke_stream_body = {
        "nuke_stream": "stream01"
    }

    # authorize 블록 병합
    payload = merge_payload(auth_hash, nuke_stream_body)
    print(f"[nukestream] 요청 페이로드: {payload}")

    # 요청 전송
    res = request_api(payload)
    print(f"[nukestream] 응답: {res}")

    return res

▸ deletestream: 설정에서 해당 스트림만 제거(실행 강제 중단 아님)
▸ deletestreamsource: 스트림 + 소스 파일 삭제 시도
▸ nuke_stream: 실행 중인 스트림을 확실히 멈추고 메모리까지 정리

 

🔷미등록 자동 정리 - no_unconfigured_streams

현재 실행 중인 스트림과 서버 설정을 비교하여, 등록되지 않았거나 설정과 다른 소스를 쓰는 스트림을 자동으로 nuke_stream 합니다.

불법 송출/오입력 정리에 효과적입니다.

def test_no_unconfigured_streams(auth_hash: str) -> dict:
    """
    no_unconfigured_streams 실습용 함수.
    - 현재 설정되지 않은 스트림 목록을 요청합니다.
    - 성공 시 응답 JSON을 그대로 반환합니다.

    Args:
        auth_hash (str): MD5(MD5(비번)+challenge)로 만든 최종 인증 해시

    Returns:
        dict: /api2 응답(JSON)
    """
    no_unconfigured_streams_body = {
        "no_unconfigured_streams": 1
    }

    # authorize 블록 병합
    payload = merge_payload(auth_hash, no_unconfigured_streams_body)
    print(f"[no_unconfigured_streams] 요청 페이로드: {payload}")

    # 요청 전송
    res = request_api(payload)
    print(f"[no_unconfigured_streams] 응답: {res}")

    return res

 

🔷세션 강제 종료 - stop_sessions

특정 스트림이나 프로토콜을 사용하는 시청자·입출력 세션을 즉시 강제 종료합니다.
명령이 실행되면 해당 세션의 연결이 곧바로 끊어집니다.

def test_stop_stssions(auth_hash: str) -> dict:
    """
    stop_sessions 실습용 함수.
    - 현재 활성화된 세션을 모두 종료합니다.
    - 성공 시 응답 JSON을 그대로 반환합니다.

    Args:
        auth_hash (str): MD5(MD5(비번)+challenge)로 만든 최종 인증 해시

    Returns:
        dict: /api2 응답(JSON)
    """
    stop_sessions_body = {
        "stop_sessions":{
           "rose":"RTSP"
         }
    }

    # authorize 블록 병합
    payload = merge_payload(auth_hash, stop_sessions_body)
    print(f"[stop_sessions] 요청 페이로드: {payload}")

    # 요청 전송
    res = request_api(payload)
    print(f"[stop_sessions] 응답: {res}")

    return res
    
    
##########################################
# "rose" 스트림의 "RTSP" protocol를 사용하는 session을 제거
"stop_sessions":{
   "rose":"RTSP"
 }
##########################################
# 다수 session을 제거하는 방법
# "rose" 스트림의 "RTMP" protocol을 사용하는 session을 제거
# ""(모든) 스트림의 "RTSP" protocol을 사용하는 session을 제거
"stop_sessions":{
  "rose": "RTMP",
   "" : "RTSP"
}

 

4. 조회와 모니터링 - active_streams, stats_streams

🔷 현재 활성 스트림 - active_streams

등록된 스트림뿐만 아니라, 일시적으로 생성되었거나 와일드카드 기반으로 동작하는 스트림도 현재 활성 상태라면 모두 조회됩니다.

또한, 조회 시 원하는 필드를 선택하거나 특정 스트림만 필터링할 수 있으며, 응답을 간략 또는 상세 형식으로 지정하는 것도 가능합니다.

#요청예시 : 특정 스트림 + 여러정보 요청
{
  "active_streams": {
    "fields": ["clients", "upbytes", "downbytes"], # 요청할 필드 선택
    "streams": ["stream1", "stream2"], # 특정 스트림만 조회
    "longform": false # (기본값) 간략한 응답 형식, true: 상세정보 응답
  }
}

#특정 정보 요청 (배열 형태)
{
  "active_streams": [
    "clients",  # 현재 연결된 클라이언트 수 (뷰어+입력+출력)
    "viewers",  # 현재 뷰어 수
    "inputs",   # 현재 입력 개수
    "outputs"   # 현재 출력 개수
  ]
}

#정보 요청 참고
{
  "active_streams": [
    #Array of string values of stream properties that are to be retrieved. 
    #Possible options are:
    "clients", #Current count of connected clients (viewers+inputs+outputs)
    "lastms", #Newest/last (currently) available timestamp in milliseconds
    "firstms", #Oldest/first requestable timestamp in milliseconds
    "viewers", #Current viewers count
    "inputs", #Current inputs count
    "outputs", #Current outputs count
    "views", #Total viewer session count since stream start
    "viewseconds", #Total sum of seconds watched by viewers since stream start
    "upbytes", #Total bytes uploaded since stream start
    "downbytes", #Total bytes downloaded since stream start
    "packsent", #Total packets sent since stream start
    "packloss", #Total packets lost since stream start
    "packretrans", #Total packets retransmitted since stream start
    "zerounix", #Unix time in seconds of timestamp epoch (zero point), if known
    "health", #Stream health object, identical to payload of STREAM_BUFFER health data
    "tracks", #Count of currently valid tracks in this stream
    "status" #Current stream status in human readable format
  ]
}

# 응답 JSON 예시 : 기본
{
  "LTS": 1,
  "active_streams": [
    "stream02"
  ],
  "authorize": {
    "status": "OK"
  }
}

#응답 JSON 예시 :  특정 정보 요청 (배열 형태)
{
  "LTS": 1,
  "active_streams": {
    "stream02": [
      2,
      1,
      1,
      0
    ]
  },
  "authorize": {
    "status": "OK"
  }
}

 

🔷 최근 10분 통계 - stats_streams

stats_streams API는 최근 약 10분 이내에 활성화된 이력이 있는 스트림 목록을 반환합니다.

현재 실행 중이지 않더라도, 10분 내에 한 번이라도 동작한 스트림이라면 결과에 포함됩니다.

오탈자에 주의하세요. 요청 키는 반드시 **stats_streams**이며, state_streams가 아닙니다.

#특정 정보 요청
{
  "stats_streams": [
    "clients", // 현재 연결된 클라이언트 수 (뷰어+입력+출력)
    "lastms"   // 현재 라이브 스트림의 최신 타임스탬프 (VoD는 항상 0)
  ]
}


#응답 JSON 예시 : 기본
{
  "LTS": 1,
  "authorize": {
    "status": "OK"
  },
  "stats_streams": [
    "",
    "stream02"
  ]
}

#응답 JSON 예시 : 특정 정보 요청
{
  "LTS": 1,
  "authorize": {
    "status": "OK"
  },
  "stats_streams": {
    "": [
      2,
      -1
    ]
  }
}

 

✔ 마무리

MistServer의 스트림 관리 API는 단순히 스트림을 등록·삭제하는 수준을 넘어, 운영 자동화와 서비스 안정성을 동시에 달성할 수 있는 강력한 도구입니다.

실무에서는 다음과 같은 방식으로 API를 결합해 운영 효율을 크게 높일 수 있습니다.

 

🔸 이상 감지와 자가 복구:

active_streams로 스트림 상태를 주기적으로 확인하고, 이상이 발견되면 stop_sessions나 nuke_stream으로 즉시 차단합니다.

또한, no_unconfigured_streams를 크론에 등록해 설정에 없는 유령 스트림을 자동으로 정리할 수 있습니다.

🔸 품질과 대역폭 모니터링:

upbytes와 downbytes 차이를 주기적으로 수집해 트래픽 변화를 파악하고, 시청자 급증 시 오토스케일이나 트랜스코딩 정책을 자동으로 조정합니다.

🔸 안전한 구성 변경:

addstream으로 필요한 스트림만 부분 업데이트하고, 문제가 발생하면 해당 스트림만 롤백해 전체 서비스 중단 없이 안정적으로 운영합니다.

즉, API를 정기 모니터링, 자동 조치, 점진적 변경에 결합하면 무중단 안정 운영이 가능합니다.

 

 

👉 관련 글 링크

1. [MistServer] 설치 실습 - Ubuntu에 직접 설치 vs Docker 컨테이너 실행

2. [MistServer] 설치 후 필수 기본 설정 가이드 (v3.8 기준)

3. [MistServer] Input 실습: 다양한 입력 방식 설정과 스트림 결과 확인

4. [MistServer] Push 기능을 이용한 스트림 녹화 설정

5. [MistServer] DVR(Time-Shift) 설정과 재생 실습

6. [MistServer] OBS Studio와 MistServer 3.8을 이용한 라이브 방송 실습(RTMP,WHIP)

7. [MistServer] API 인증 방식과 HTTP·WebSocket API 비교

8. [MistServer] 스트림 관리 API 실습 (Python)

 

 

 

반응형

※ 게시된 글 및 이미지 중 일부는 AI 도구의 도움을 받아 생성되거나 다듬어졌습니다.