Upload schedule_api_client.py

schedule_api_client.py

@@ -0,0 +1,337 @@
+"""
+与 app.py 一致的票务排程 API（Token、影厅排片），供多页面复用，避免 import app 时执行整站 UI。
+"""
+import json
+import os
+import time
+
+import pandas as pd
+import requests
+from dotenv import load_dotenv
+
+
+class _NoopStreamlit:
+    @staticmethod
+    def error(*args, **kwargs):
+        return None
+
+    @staticmethod
+    def toast(*args, **kwargs):
+        return None
+
+    @staticmethod
+    def cache_data(*args, **kwargs):
+        def _decorator(func):
+            return func
+
+        return _decorator
+
+
+def _resolve_streamlit():
+    """
+    - Streamlit 页面内：保留原能力（toast/cache_data）
+    - 非 Streamlit 运行（如 Flask/FastAPI）：使用 no-op，避免无运行时警告
+    """
+    try:
+        import streamlit as _st
+        from streamlit.runtime.scriptrunner import get_script_run_ctx
+
+        if get_script_run_ctx(suppress_warning=True) is None:
+            return _NoopStreamlit()
+        return _st
+    except Exception:
+        return _NoopStreamlit()
+
+
+st = _resolve_streamlit()
+
+load_dotenv()
+
+TOKEN_FILE = "token_data.json"
+CINEMA_ID = os.getenv("CINEMA_ID")
+
+
+def load_token():
+    if os.path.exists(TOKEN_FILE):
+        try:
+            with open(TOKEN_FILE, "r", encoding="utf-8") as f:
+                return json.load(f)
+        except (json.JSONDecodeError, FileNotFoundError):
+            return None
+    return None
+
+
+def save_token(token_data):
+    try:
+        with open(TOKEN_FILE, "w", encoding="utf-8") as f:
+            json.dump(token_data, f, ensure_ascii=False, indent=4)
+        return True
+    except Exception as e:
+        st.error(f"保存Token失败: {e}")
+        return False
+
+
+def login_and_get_token():
+    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]):
+        st.error("登录失败：未配置用户名、密码或影院编码环境变量。")
+        return None
+
+    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):
+            st.error(f"登录步骤 1 失败，未能获取 Session Token。状态码: {response_login.status_code}")
+            return None
+
+        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"]
+            if save_token(token_data):
+                st.toast("登录成功，已获取并保存新 Token！", icon="🔑")
+            return token_data
+        st.error(f"登录步骤 2 失败，未能从 JSON 中提取 Token。响应: {user_info.get('msg')}")
+        return None
+
+    except requests.exceptions.RequestException as e:
+        st.error(f"登录请求过程中发生网络错误: {e}")
+        return None
+
+
+def fetch_hall_info(token):
+    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"}
+    response = requests.get(url, params=params, headers=headers, timeout=10)
+    response.raise_for_status()
+    data = response.json()
+    if data.get("code") == 1 and data.get("data"):
+        return {item["hallId"]: item["seatNum"] for item in data["data"]}
+    raise Exception(f"获取影厅信息失败: {data.get('msg', '未知错误')}")
+
+
+def fetch_schedule_data(token, show_date):
+    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"}
+    response = requests.get(url, params=params, headers=headers, timeout=15)
+    response.raise_for_status()
+    data = response.json()
+    if data.get("code") == 1:
+        return data.get("data", [])
+    if data.get("code") == 500:
+        raise ValueError("Token 可能已失效")
+    raise Exception(f"获取排片数据失败: {data.get('msg', '未知错误')}")
+
+
+def get_api_data_with_token_management(show_date):
+    token_data = load_token()
+    token = token_data.get("token") if token_data else None
+    if not token:
+        token_data = login_and_get_token()
+        if not token_data:
+            return None, None
+        token = token_data.get("token")
+
+    try:
+        schedule_list = fetch_schedule_data(token, show_date)
+        hall_seat_map = fetch_hall_info(token)
+        return schedule_list, hall_seat_map
+    except ValueError:
+        st.toast("Token 已失效，正在尝试重新登录并重试...", icon="🔄")
+        token_data = login_and_get_token()
+        if not token_data:
+            return None, None
+        token = token_data.get("token")
+        try:
+            schedule_list = fetch_schedule_data(token, show_date)
+            hall_seat_map = fetch_hall_info(token)
+            return schedule_list, hall_seat_map
+        except Exception as e:
+            st.error(f"重试获取数据失败: {e}")
+            return None, None
+    except Exception as e:
+        st.error(f"获取 API 数据时发生错误: {e}")
+        return None, None
+
+
+@st.cache_data(show_spinner=False, ttl=600)
+def fetch_canonical_movie_names(token, date_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 as e:
+        print(f"获取标准电影名称失败: {e}")
+    return []
+
+
+def clean_movie_title(raw_title, canonical_names=None):
+    if not isinstance(raw_title, str):
+        return raw_title
+
+    base_name = None
+
+    if canonical_names:
+        sorted_names = sorted(canonical_names, key=len, reverse=True)
+        for name in sorted_names:
+            if name in raw_title:
+                base_name = name
+                break
+
+    if not base_name:
+        base_name = raw_title.split(" ", 1)[0]
+
+    raw_upper = raw_title.upper()
+    suffix = ""
+
+    if "HDR LED" in raw_upper:
+        suffix = "（HDR LED）"
+    elif "CINITY" in raw_upper:
+        suffix = "（CINITY）"
+    elif "杜比" in raw_upper or "DOLBY" in raw_upper:
+        suffix = "（杜比视界）"
+    elif "IMAX" in raw_upper:
+        suffix = "（数字IMAX3D）" if "3D" in raw_upper else "（数字IMAX）"
+    elif "巨幕" in raw_upper:
+        suffix = "（中国巨幕立体）" if "立体" in raw_upper else "（中国巨幕）"
+    elif "3D" in raw_upper:
+        suffix = "（数字3D）"
+
+    if suffix and suffix not in base_name:
+        return f"{base_name}{suffix}"
+
+    return base_name
+
+
+def get_valid_token(force_refresh=False):
+    token_data = None if force_refresh else load_token()
+    if not token_data:
+        token_data = login_and_get_token()
+    if not token_data:
+        return None
+    return token_data.get("token")
+
+
+def fetch_schedule_api_bundle(show_date):
+    """
+    一次性获取排程相关 API 原始数据：
+    - getHallShowInfo（场次列表）
+    - getShowHallInfo（影厅座位映射）
+    - movieSellGross（标准影片名称）
+    """
+    schedule_list, hall_seat_map = get_api_data_with_token_management(show_date)
+    if schedule_list is None or hall_seat_map is None:
+        return None
+
+    token_data = load_token()
+    token = token_data.get("token") if token_data else None
+    canonical_names = fetch_canonical_movie_names(token, show_date) if token else []
+
+    return {
+        "show_date": show_date,
+        "token": token,
+        "schedule_list": schedule_list,
+        "hall_seat_map": hall_seat_map,
+        "canonical_names": canonical_names,
+    }
+
+
+def process_schedule_dataframe(schedule_list, hall_seat_map, canonical_names=None):
+    """将排程 API 原始数据整理成便于展示的表格。"""
+    if not schedule_list:
+        return pd.DataFrame()
+
+    df = pd.DataFrame(schedule_list)
+    if df.empty:
+        return pd.DataFrame()
+
+    df["座位数"] = df["hallId"].map(hall_seat_map or {}).fillna(0).astype(int)
+    df.rename(
+        columns={
+            "movieName": "影片名称",
+            "showStartTime": "放映时间",
+            "soldBoxOffice": "总收入",
+            "soldTicketNum": "总人次",
+            "hallName": "影厅名称",
+            "showEndTime": "散场时间",
+        },
+        inplace=True,
+    )
+
+    if "影片名称" in df.columns:
+        df["影片名称_清洗后"] = df["影片名称"].apply(
+            lambda x: clean_movie_title(x, canonical_names)
+        )
+
+    required_cols = [
+        "影片名称",
+        "影片名称_清洗后",
+        "放映时间",
+        "散场时间",
+        "影厅名称",
+        "座位数",
+        "总收入",
+        "总人次",
+    ]
+    for col in required_cols:
+        if col not in df.columns:
+            df[col] = None
+
+    df = df[required_cols]
+    for col in ["座位数", "总收入", "总人次"]:
+        df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0)
+
+    return df