cinema_api_client.py

import json
import os
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import requests
from dotenv import load_dotenv


load_dotenv()

ROOT_DIR = Path(__file__).resolve().parent
TOKEN_FILE = ROOT_DIR / "token_data.json"
CINEMA_ID = os.getenv("CINEMA_ID")
DEFAULT_SELF_CINEMA_ID = "44001291"
SCHEDULE_API_TIMEOUT_SECONDS = 30
SCHEDULE_API_RETRY_ATTEMPTS = 2
API_BASE_URL = "https://cawapi.yinghezhong.com"
COMMON_HEADERS = {
    "Host": "cawapi.yinghezhong.com",
    "Accept": "*/*",
    "Origin": "https://caw.yinghezhong.com",
    "Connection": "keep-alive",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Site": "same-site",
    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
    "Referer": "https://caw.yinghezhong.com/",
    "Sec-Fetch-Dest": "empty",
    "Accept-Language": "zh-CN,zh-Hans;q=0.9",
}


class RetryableAPIError(RuntimeError):
    """适合重试的接口异常。"""


def load_token() -> Optional[dict]:
    if not TOKEN_FILE.exists():
        return None
    try:
        return json.loads(TOKEN_FILE.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return None


def save_token(token_data: dict) -> bool:
    try:
        TOKEN_FILE.write_text(json.dumps(token_data, ensure_ascii=False, indent=2), encoding="utf-8")
        return True
    except OSError:
        return False


def login_and_get_token() -> dict:
    username = os.getenv("CINEMA_USERNAME")
    password = os.getenv("CINEMA_PASSWORD")
    res_code = os.getenv("CINEMA_RES_CODE")
    device_id = os.getenv("CINEMA_DEVICE_ID")

    if not all([username, password, res_code]):
        raise RuntimeError("未配置 CINEMA_USERNAME / CINEMA_PASSWORD / CINEMA_RES_CODE。")

    session = requests.Session()
    session.headers.update(
        {
            "Host": "app.bi.piao51.cn",
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
        }
    )

    login_url = "https://app.bi.piao51.cn/cinema-app/credential/login.action"
    login_headers = {
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "https://app.bi.piao51.cn",
    }
    login_data = {
        "username": username,
        "password": password,
        "type": "1",
        "resCode": res_code,
        "deviceid": device_id,
        "dtype": "ios",
    }

    try:
        response_login = session.post(
            login_url,
            headers=login_headers,
            data=login_data,
            allow_redirects=False,
            timeout=15,
        )
        if not (300 <= response_login.status_code < 400 and "token" in session.cookies):
            raise RetryableAPIError(f"票务登录失败，状态码：{response_login.status_code}")

        user_info_url = "https://app.bi.piao51.cn/cinema-app/security/logined.action"
        response_user_info = session.get(user_info_url, timeout=10)
        response_user_info.raise_for_status()
        user_info = response_user_info.json()

        if user_info.get("success") and user_info.get("data", {}).get("token"):
            token_data = user_info["data"]
            save_token(token_data)
            return token_data
        raise RetryableAPIError(f"票务登录未获取到 Token：{user_info.get('msg')}")
    except RetryableAPIError:
        raise
    except requests.exceptions.RequestException as exc:
        raise RetryableAPIError(f"票务登录接口异常：{exc}") from exc
    except Exception as exc:
        raise RetryableAPIError(f"票务登录处理异常：{exc}") from exc


def get_valid_cinema_token() -> str:
    token_data = load_token()
    token = token_data.get("token") if token_data else None
    if token:
        return token
    token_data = login_and_get_token()
    token = token_data.get("token") if token_data else None
    if not token:
        raise RetryableAPIError("未获取到票务 Token。")
    return token


def fetch_hall_info(token: str) -> Dict[str, int]:
    url = "https://cawapi.yinghezhong.com/showInfo/getShowHallInfo"
    params = {"token": token, "_": int(time.time() * 1000)}
    headers = {"Origin": "https://caw.yinghezhong.com", "User-Agent": "Mozilla/5.0"}

    try:
        response = requests.get(url, params=params, headers=headers, timeout=10)
        response.raise_for_status()
        payload = response.json()
    except requests.exceptions.RequestException as exc:
        raise RetryableAPIError(f"获取影厅信息接口异常：{exc}") from exc
    except Exception as exc:
        raise RetryableAPIError(f"获取影厅信息处理异常：{exc}") from exc

    if payload.get("code") == 1 and payload.get("data") is not None:
        return {str(item["hallId"]): int(item.get("seatNum") or 0) for item in payload["data"]}
    if payload.get("code") == 500:
        raise ValueError("Token 可能已失效")
    raise RetryableAPIError(f"获取影厅信息失败：{payload.get('msg', '未知错误')}")


def fetch_schedule_data(token: str, show_date: str) -> List[dict]:
    url = "https://cawapi.yinghezhong.com/showInfo/getHallShowInfo"
    params = {"showDate": show_date, "token": token, "_": int(time.time() * 1000)}
    headers = {"Origin": "https://caw.yinghezhong.com", "User-Agent": "Mozilla/5.0"}
    last_request_error = None

    for attempt in range(1, SCHEDULE_API_RETRY_ATTEMPTS + 1):
        try:
            response = requests.get(
                url,
                params=params,
                headers=headers,
                timeout=SCHEDULE_API_TIMEOUT_SECONDS,
            )
            response.raise_for_status()
            payload = response.json()
        except requests.exceptions.RequestException as exc:
            last_request_error = exc
            if attempt < SCHEDULE_API_RETRY_ATTEMPTS:
                time.sleep(2)
                continue
            raise RetryableAPIError(f"获取 {show_date} 排片接口异常：{exc}") from exc
        except Exception as exc:
            raise RetryableAPIError(f"获取 {show_date} 排片处理异常：{exc}") from exc

        if payload.get("code") == 1:
            return payload.get("data", [])
        if payload.get("code") == 500:
            raise ValueError("Token 可能已失效")
        if attempt < SCHEDULE_API_RETRY_ATTEMPTS:
            time.sleep(2)
            continue
        raise RetryableAPIError(f"获取 {show_date} 排片失败：{payload.get('msg', '未知错误')}")

    raise RetryableAPIError(f"获取 {show_date} 排片接口异常：{last_request_error}")


def fetch_schedule_data_by_cinema(token: str, cinema_id: str, show_date: str) -> List[dict]:
    cinema_id = str(cinema_id or "").strip()
    self_cinema_id = (CINEMA_ID or "").strip() or DEFAULT_SELF_CINEMA_ID
    if cinema_id == self_cinema_id:
        return fetch_schedule_data(token, show_date)

    url = f"{API_BASE_URL}/competition/play/showInfoList"
    params = {
        "date": datetime.strptime(show_date, "%Y-%m-%d").strftime("%Y/%m/%d"),
        "cinemaNum": cinema_id,
        "token": token,
        "_": str(int(time.time() * 1000)),
    }
    last_request_error = None

    for attempt in range(1, SCHEDULE_API_RETRY_ATTEMPTS + 1):
        try:
            response = requests.get(
                url,
                params=params,
                headers=COMMON_HEADERS,
                timeout=SCHEDULE_API_TIMEOUT_SECONDS,
            )
            response.raise_for_status()
            payload = response.json()
        except requests.exceptions.RequestException as exc:
            last_request_error = exc
            if attempt < SCHEDULE_API_RETRY_ATTEMPTS:
                time.sleep(2)
                continue
            raise RetryableAPIError(f"获取影院 {cinema_id} 在 {show_date} 的竞对排片接口异常：{exc}") from exc
        except Exception as exc:
            raise RetryableAPIError(f"获取影院 {cinema_id} 在 {show_date} 的竞对排片处理异常：{exc}") from exc

        if payload.get("code") == 1:
            return payload.get("data", []) or []
        if payload.get("code") == 500:
            raise ValueError("Token 可能已失效")
        if attempt < SCHEDULE_API_RETRY_ATTEMPTS:
            time.sleep(2)
            continue
        raise RetryableAPIError(
            f"获取影院 {cinema_id} 在 {show_date} 的竞对排片失败：{payload.get('msg', '未知错误')}"
        )

    raise RetryableAPIError(f"获取影院 {cinema_id} 在 {show_date} 的竞对排片接口异常：{last_request_error}")


def fetch_available_movie_info(token: str, show_date: str) -> List[dict]:
    url = "https://cawapi.yinghezhong.com/show/getMovieInfo"
    params = {
        "showDate": show_date,
        "token": token,
        "_": int(time.time() * 1000),
    }
    headers = {
        "Origin": "https://caw.yinghezhong.com",
        "Referer": "https://caw.yinghezhong.com/",
        "User-Agent": "Mozilla/5.0",
    }

    try:
        response = requests.get(url, params=params, headers=headers, timeout=15)
        response.raise_for_status()
        payload = response.json()
    except requests.exceptions.RequestException as exc:
        raise RetryableAPIError(f"获取可排片池接口异常：{exc}") from exc
    except Exception as exc:
        raise RetryableAPIError(f"获取可排片池处理异常：{exc}") from exc

    if payload.get("code") == 1:
        return payload.get("data", [])
    if payload.get("code") == 500:
        raise ValueError("Token 可能已失效")
    raise RetryableAPIError(f"获取可排片池失败：{payload.get('msg', '未知错误')}")


def fetch_realtime_box_office(token: str, date_str: str) -> dict:
    url = "https://app.bi.piao51.cn/cinema-app/market/realtimeDailyBoxOffice.action"
    params = {
        "qTime": date_str,
        "token": token,
    }
    headers = {
        "Host": "app.bi.piao51.cn",
        "X-Requested-With": "XMLHttpRequest",
        "jwt": "0",
        "Accept": "application/json, text/javascript, */*; q=0.01",
        "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
    }

    try:
        response = requests.get(url, params=params, headers=headers, timeout=15)
        response.raise_for_status()
        payload = response.json()
    except requests.exceptions.RequestException as exc:
        raise RetryableAPIError(f"获取 {date_str} 全国大盘接口异常：{exc}") from exc
    except Exception as exc:
        raise RetryableAPIError(f"获取 {date_str} 全国大盘处理异常：{exc}") from exc

    if payload.get("code") == "A00000":
        return payload.get("results", {}) or {}
    if "login" in response.text or payload.get("code") in {500, "500"}:
        raise ValueError("Token 可能已失效")
    raise RetryableAPIError(f"获取 {date_str} 全国大盘失败：{payload.get('msg', '未知错误')}")


def fetch_canonical_movie_names(token: str, date_str: str) -> List[str]:
    if not CINEMA_ID:
        return []

    url = "https://app.bi.piao51.cn/cinema-app/mycinema/movieSellGross.action"
    params = {
        "token": token,
        "startDate": date_str,
        "endDate": date_str,
        "dateType": "day",
        "cinemaId": CINEMA_ID,
    }
    headers = {
        "Host": "app.bi.piao51.cn",
        "X-Requested-With": "XMLHttpRequest",
        "jwt": "0",
        "Accept": "application/json, text/javascript, */*; q=0.01",
        "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148",
    }

    try:
        response = requests.get(url, params=params, headers=headers, timeout=10)
        response.raise_for_status()
        data = response.json()
        if data.get("code") == "A00000" and data.get("results"):
            return [item["movieName"] for item in data["results"] if item.get("movieName") and item["movieName"] != "总计"]
    except Exception:
        return []
    return []


def get_schedule_and_hall_info(show_date: str) -> Tuple[List[dict], Dict[str, int], str]:
    token = get_valid_cinema_token()
    try:
        schedule = fetch_schedule_data(token, show_date)
        hall_map = fetch_hall_info(token)
        return schedule, hall_map, token
    except ValueError:
        token_data = login_and_get_token()
        token = token_data.get("token") if token_data else None
        if not token:
            raise RetryableAPIError("重新登录后仍未获取到票务 Token。")
        schedule = fetch_schedule_data(token, show_date)
        hall_map = fetch_hall_info(token)
        return schedule, hall_map, token


def get_schedule_by_cinema_with_token_management(cinema_id: str, show_date: str) -> Tuple[List[dict], str]:
    token = get_valid_cinema_token()
    try:
        return fetch_schedule_data_by_cinema(token, cinema_id, show_date), token
    except ValueError:
        token_data = login_and_get_token()
        token = token_data.get("token") if token_data else None
        if not token:
            raise RetryableAPIError("重新登录后仍未获取到票务 Token。")
        return fetch_schedule_data_by_cinema(token, cinema_id, show_date), token


def get_available_movies_with_token_management(show_date: str) -> Tuple[List[dict], str]:
    token = get_valid_cinema_token()
    try:
        return fetch_available_movie_info(token, show_date), token
    except ValueError:
        token_data = login_and_get_token()
        token = token_data.get("token") if token_data else None
        if not token:
            raise RetryableAPIError("重新登录后仍未获取到票务 Token。")
        return fetch_available_movie_info(token, show_date), token


def get_realtime_box_office_with_token_management(date_str: str) -> Tuple[dict, str]:
    token = get_valid_cinema_token()
    try:
        return fetch_realtime_box_office(token, date_str), token
    except ValueError:
        token_data = login_and_get_token()
        token = token_data.get("token") if token_data else None
        if not token:
            raise RetryableAPIError("重新登录后仍未获取到票务 Token。")
        return fetch_realtime_box_office(token, date_str), token