Upload 🆕 新API影片映出日累计报表测试.py

pages/🆕 新API影片映出日累计报表测试.py

@@ -0,0 +1,541 @@
+"""新 API（center.hengdianfilm.com / cinema/movieshow/page）测试页面。
+
+复用 app.py 的「影片映出日累计报表」生成逻辑，但数据来源切换为：
+1. 直接调用新 API 拉取（需 Bearer Token）
+2. 手动粘贴 API 返回 JSON 文本
+
+最终生成同样格式的报表，并支持下载 XLSX。
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import re
+from datetime import date as dt_date, datetime, time as dt_time, timedelta
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import requests
+import streamlit as st
+
+
+# ---------------------------------------------------------------------------
+# 页面配置
+# ---------------------------------------------------------------------------
+st.set_page_config(layout="wide", page_title="🆕 新 API 影片映出日累计报表测试")
+st.title("🆕 新 API 影片映出日累计报表测试")
+st.caption(
+    "测试 `https://center.hengdianfilm.com/cinema/movieshow/page` 接口，"
+    "数据可由 API 抓取或直接粘贴 JSON 返回。生成与 `app.py` 一致格式的「影片映出日累计报表」。"
+)
+
+
+# ---------------------------------------------------------------------------
+# 常量与路径
+# ---------------------------------------------------------------------------
+ROOT_DIR = Path(__file__).resolve().parent.parent
+CINEMA_CACHE_DIR = ROOT_DIR / "cinema_cache"
+MOVIE_NUM_NAME_MAP_FILE = CINEMA_CACHE_DIR / "movie_num_name_map.json"
+NEW_API_URL = "https://center.hengdianfilm.com/cinema/movieshow/page"
+DEFAULT_PAGE_SIZE = 200
+DEFAULT_BEARER_TOKEN = ""
+
+
+# ---------------------------------------------------------------------------
+# 工具函数（与 app.py 保持一致；为保持本页独立，做了最小复制）
+# ---------------------------------------------------------------------------
+def _normalize_movie_num_key(movie_num) -> str:
+    return re.sub(r"[^A-Z0-9]", "", str(movie_num or "").strip().upper())
+
+
+@st.cache_data(show_spinner=False, ttl=120)
+def _load_movie_num_name_map() -> dict:
+    """读取 cinema_cache/movie_num_name_map.json，返回 {规范化 movieNum: 官方名称}。"""
+    if not MOVIE_NUM_NAME_MAP_FILE.exists():
+        return {}
+    try:
+        with open(MOVIE_NUM_NAME_MAP_FILE, "r", encoding="utf-8") as f:
+            payload = json.load(f)
+    except Exception:
+        return {}
+    movie_num_map = payload.get("movie_num_map", {})
+    if not isinstance(movie_num_map, dict):
+        return {}
+    result = {}
+    for movie_num, entry in movie_num_map.items():
+        key = _normalize_movie_num_key(movie_num)
+        if not key:
+            continue
+        if isinstance(entry, str):
+            name = entry.strip()
+        elif isinstance(entry, dict):
+            name = str(entry.get("official_name") or "").strip()
+        else:
+            continue
+        if name:
+            result[key] = name
+    return result
+
+
+def clean_movie_title(raw_title, canonical_names=None):
+    """与 app.py 同名函数完全一致的精简版。"""
+    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 resolve_movie_name_from_schedule_item(raw_movie_name, movie_num=None, canonical_names=None):
+    """与 app.py 同名函数一致：优先用 movieNum 映射，回退到 canonical_names / 空格分割。"""
+    if not isinstance(raw_movie_name, str):
+        return raw_movie_name
+    if movie_num:
+        key = _normalize_movie_num_key(movie_num)
+        if key:
+            official = _load_movie_num_name_map().get(key)
+            if official:
+                return official
+
+    if canonical_names:
+        for name in sorted(canonical_names, key=len, reverse=True):
+            if name in raw_movie_name:
+                return name
+    return raw_movie_name.split(" ", 1)[0]
+
+
+def normalize_report_time_value(value):
+    if pd.isna(value):
+        return None
+    if isinstance(value, datetime):
+        return value.time().replace(second=0, microsecond=0)
+    if isinstance(value, dt_time):
+        return value.replace(second=0, microsecond=0)
+    parsed = pd.to_datetime(str(value).strip(), errors="coerce")
+    if pd.isna(parsed):
+        return None
+    return parsed.time().replace(second=0, microsecond=0)
+
+
+def infer_daily_report_date(date_series, fallback_date_str=None):
+    parsed = pd.to_datetime(date_series, errors="coerce").dropna()
+    if not parsed.empty:
+        return parsed.dt.date.mode().iloc[0]
+    if fallback_date_str:
+        fb = pd.to_datetime(fallback_date_str, errors="coerce")
+        if pd.notna(fb):
+            return fb.date()
+    return None
+
+
+def build_daily_report_from_source_df(source_df, selected_date_str=None, canonical_names=None):
+    """与 app.py 中 build_daily_report_from_source_df 一致，返回 (df, display_date)。"""
+    if source_df is None or source_df.empty:
+        return pd.DataFrame(), infer_daily_report_date(pd.Series(dtype=object), selected_date_str)
+
+    required_cols = ["影片名称", "放映时间", "影厅名称", "总人次", "座位数"]
+    missing = [c for c in required_cols if c not in source_df.columns]
+    if missing:
+        st.error(f"数据缺少必要列: {', '.join(missing)}")
+        return pd.DataFrame(), None
+
+    df = source_df.copy()
+    if "放映日期" not in df.columns:
+        df["放映日期"] = selected_date_str
+
+    display_date = infer_daily_report_date(df["放映日期"], selected_date_str)
+    display_date_str = display_date.strftime("%Y-%m-%d") if display_date else selected_date_str
+
+    df["影片名称"] = df["影片名称"].astype(str).str.strip()
+    df["影厅名称"] = df["影厅名称"].fillna("未知影厅").astype(str).str.strip()
+    df["总人次"] = pd.to_numeric(df["总人次"], errors="coerce").fillna(0).round().astype(int)
+    df["座位数"] = pd.to_numeric(df["座位数"], errors="coerce").fillna(0).round().astype(int)
+    df["_放映时间对象"] = df["放映时间"].apply(normalize_report_time_value)
+    df["放映日期"] = pd.to_datetime(df["放映日期"], errors="coerce").dt.strftime("%Y-%m-%d")
+    if display_date_str:
+        df["放映日期"] = df["放映日期"].fillna(display_date_str)
+
+    df.dropna(subset=["影片名称", "_放映时间对象"], inplace=True)
+    df = df[df["影片名称"].ne("") & df["影片名称"].ne("nan")].copy()
+    df = df[df["总人次"] > 0].copy()
+
+    if df.empty:
+        st.info("所有场次的观影人数均为 0，没有可显示的数据。")
+        return pd.DataFrame(), display_date
+
+    if "movieNum" in df.columns:
+        df["影片"] = df.apply(
+            lambda row: resolve_movie_name_from_schedule_item(
+                row["影片名称"],
+                movie_num=row.get("movieNum"),
+                canonical_names=canonical_names,
+            ),
+            axis=1,
+        )
+    else:
+        df["影片"] = df["影片名称"].apply(lambda x: clean_movie_title(x, canonical_names))
+    df["影厅"] = df["影厅名称"]
+    df["人数合计"] = df["总人次"]
+
+    with np.errstate(divide="ignore", invalid="ignore"):
+        df["上座率%"] = np.divide(df["人数合计"], df["座位数"]) * 100
+        df["上座率%"] = df["上座率%"].replace([np.inf, -np.inf], 0).fillna(0)
+
+    df["放映时间"] = df["_放映时间对象"].apply(lambda x: x.strftime("%H:%M:%S"))
+    result = df[["影片", "放映日期", "放映时间", "影厅", "人数合计", "座位数", "上座率%"]].copy()
+    result = result.sort_values(by=["放映日期", "放映时间", "影厅", "影片"]).reset_index(drop=True)
+    return result, display_date
+
+
+# ---------------------------------------------------------------------------
+# 新 API 数据获取与解析
+# ---------------------------------------------------------------------------
+def _build_request_headers(bearer_token: str) -> dict:
+    return {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:115.0) Gecko/20100101 Firefox/115.0",
+        "Accept": "application/json, text/plain, */*",
+        "Accept-Language": "zh-CN,zh;q=0.9",
+        "Content-Type": "application/json;charset=utf-8",
+        "Authorization": f"Bearer {bearer_token.strip()}",
+        "TENANT-ID": "1",
+        "Channel": "4",
+        "Origin": "https://center.hengdianfilm.com",
+        "Referer": "https://center.hengdianfilm.com/",
+        "Connection": "keep-alive",
+    }
+
+
+def fetch_movieshow_page(bearer_token: str, show_date: str, current: int = 1, size: int = DEFAULT_PAGE_SIZE):
+    """单次拉取一页排片数据。返回原始 JSON dict。"""
+    payload = {
+        "size": size,
+        "current": current,
+        "entity": {"cinemaMovieShowDate": show_date},
+    }
+    resp = requests.post(
+        NEW_API_URL,
+        headers=_build_request_headers(bearer_token),
+        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
+        timeout=20,
+    )
+    resp.raise_for_status()
+    return resp.json()
+
+
+def fetch_all_movieshow(bearer_token: str, show_date: str, size: int = DEFAULT_PAGE_SIZE) -> list:
+    """自动翻页拉取所有场次记录。"""
+    all_records: list = []
+    current = 1
+    while True:
+        data = fetch_movieshow_page(bearer_token, show_date, current=current, size=size)
+        if data.get("code") != 0:
+            raise RuntimeError(f"接口返回失败：{data.get('msg') or data}")
+        body = data.get("data") or {}
+        records = body.get("records") or []
+        all_records.extend(records)
+        total = int(body.get("total") or 0)
+        if len(all_records) >= total or not records:
+            break
+        current += 1
+        if current > 50:  # 安全阈值
+            break
+    return all_records
+
+
+def parse_pasted_response(raw_text: str) -> list:
+    """解析用户粘贴的 JSON 响应文本，返回 records 列表。"""
+    text = (raw_text or "").strip()
+    if not text:
+        raise ValueError("粘贴内容为空。")
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"JSON 解析失败：{exc}") from exc
+
+    # 用户既可能粘贴整个响应，也可能直接粘贴 records 列表
+    if isinstance(payload, list):
+        return payload
+    if not isinstance(payload, dict):
+        raise ValueError("无法识别的 JSON 结构，期望对象或数组。")
+
+    if "data" in payload and isinstance(payload["data"], dict):
+        records = payload["data"].get("records")
+        if isinstance(records, list):
+            return records
+    if "records" in payload and isinstance(payload["records"], list):
+        return payload["records"]
+    raise ValueError("未在 JSON 中找到 records 数组。请粘贴接口完整返回或 records 列表。")
+
+
+def records_to_source_df(records: list) -> pd.DataFrame:
+    """将新 API 返回的场次列表转换成 build_daily_report_from_source_df 期望的列结构。"""
+    if not records:
+        return pd.DataFrame()
+    rows = []
+    for item in records:
+        if not isinstance(item, dict):
+            continue
+        rows.append(
+            {
+                "影片名称": item.get("cinemaMovieName") or "",
+                "放映时间": item.get("cinemaMovieShowStartTime") or "",
+                "影厅名称": item.get("cinemaHallName") or "",
+                "总人次": item.get("cinemaMovieShowSoldNum") or 0,
+                "座位数": item.get("cinemaHallSeatNum") or 0,
+                "放映日期": item.get("cinemaMovieShowDate") or "",
+                "movieNum": item.get("cinemaMovieNum") or "",
+            }
+        )
+    return pd.DataFrame(rows)
+
+
+# ---------------------------------------------------------------------------
+# Session State 初始化
+# ---------------------------------------------------------------------------
+def _init_state():
+    defaults = {
+        "new_api_bearer_token": os.getenv("HENGDIAN_CENTER_TOKEN", DEFAULT_BEARER_TOKEN),
+        "new_api_query_date": dt_date.today() + timedelta(days=1),
+        "new_api_records": [],
+        "new_api_report_df": pd.DataFrame(),
+        "new_api_display_date": None,
+        "new_api_data_source": "",
+        "new_api_paste_text": "",
+    }
+    for key, value in defaults.items():
+        st.session_state.setdefault(key, value)
+
+
+_init_state()
+
+
+# ---------------------------------------------------------------------------
+# UI
+# ---------------------------------------------------------------------------
+with st.expander("📘 接口说明", expanded=False):
+    st.markdown(
+        f"""
+- **接口**：`POST {NEW_API_URL}`
+- **必备 Header**：`Authorization: Bearer <token>`、`TENANT-ID: 1`、`Channel: 4`、`Content-Type: application/json;charset=utf-8`
+- **请求体**：`{{"size": 100, "current": 1, "entity": {{"cinemaMovieShowDate": "YYYY-MM-DD"}}}}`
+- **关键返回字段**：
+  - `cinemaMovieName` → 影片名称（含制式）
+  - `cinemaMovieNum` → 影片编号（用于 `movie_num_name_map.json` 标准名映射）
+  - `cinemaMovieShowStartTime` → 放映开始时间
+  - `cinemaHallName` / `cinemaHallSeatNum` → 影厅名称 / 座位数
+  - `cinemaMovieShowSoldNum` → 已售人次
+        """
+    )
+
+with st.expander("🔑 如何获取 Bearer Token（浏览器开发者工具教程）", expanded=False):
+    st.markdown(
+        """
+> Token 是登录后由浏览器临时保存的身份凭证,**有效期通常只有几小时到一天**,失效后需要重新抓取。
+
+#### 方法 A:从「Network(网络)」面板抓取(推荐,最准确)
+
+1. 用 Chrome / Edge / Firefox 打开 [https://center.hengdianfilm.com/](https://center.hengdianfilm.com/) 并**正常登录**。
+2. 按 **F12**(或右键 → **检查 / Inspect**)打开开发者工具,切换到 **Network(网络)** 面板。
+3. 勾选 **Preserve log(保留日志)**,在过滤框里输入关键字 `movieshow` 或 `cinema/`,过滤出业务接口。
+4. 在网页上**点击任意会触发刷新数据的操作**(例如切换日期、点查询),让接口重新发一次请求。
+5. 在过滤结果里点击��条形如 `movieshow/page` 的请求 → 切到 **Headers(标头)** 子面板 → 找到 **Request Headers(请求标头)**。
+6. 复制 `Authorization:` 这一行后面的内容,**去掉开头的 `Bearer `**,只保留后面那串形如 `40ea061e-598e-4e6f-8f98-...` 的字符串,粘贴到下方输入框即可。
+
+#### 方法 B:从「Application / 存储」面板里直接读取
+
+1. 同样打开开发者工具,切到 **Application(应用)** 面板(Firefox 叫 **Storage / 存储**)。
+2. 左侧依次点开 **Local Storage** 和 **Cookies**,目标域名选 `https://center.hengdianfilm.com`。
+3. 在右侧的键值对里搜索关键字: `token`、`access_token`、`Authorization`、`auth`、`satoken`。
+4. 找到形如 UUID(`xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx`)的值,复制粘贴到下方输入框。
+
+#### 方法 C:Console(控制台)一键提取
+
+打开 **Console(控制台)** 面板,粘贴下面任意一行回车,即可直接打印出 token:
+
+```js
+// 如果存在 localStorage 里
+copy(localStorage.getItem('Authorization') || localStorage.getItem('token') || localStorage.getItem('access_token'));
+```
+
+```js
+// 一键查找所有可能键
+Object.keys(localStorage).filter(k => /token|auth/i.test(k)).forEach(k => console.log(k, '=', localStorage.getItem(k)));
+```
+
+执行 `copy(...)` 后 token 会被复制到剪贴板,直接粘贴即可。
+
+#### 常见问题
+
+- **过期 / 401 Unauthorized** → 重新登录 hengdianfilm 后台,再次按上面步骤抓取。
+- **找不到 `Authorization` 头** → 确认你已经登录,并且过滤的是业务接口(URL 含 `/cinema/`),而不是登录接口本身。
+- **复制时多了 `Bearer ` 前缀** → 输入框里**不要**带 `Bearer `,代码里会自动拼接。
+        """
+    )
+
+input_tab_api, input_tab_paste = st.tabs(["🌐 API 拉取", "📋 粘贴 JSON"])
+
+
+with input_tab_api:
+    col_token, col_date = st.columns([3, 2])
+    with col_token:
+        bearer_input = st.text_input(
+            "Bearer Token",
+            value=st.session_state.new_api_bearer_token,
+            help="对应 curl 中的 `Authorization: Bearer <token>`。",
+            key="new_api_bearer_input",
+        )
+    with col_date:
+        query_date = st.date_input(
+            "查询排片日期",
+            value=st.session_state.new_api_query_date,
+            key="new_api_date_input",
+        )
+
+    page_size = st.slider("每页大小（自动翻页直到取完）", min_value=50, max_value=500, value=DEFAULT_PAGE_SIZE, step=50)
+
+    if st.button("🫵 拉取并生成报表", type="primary", use_container_width=False):
+        token = (bearer_input or "").strip()
+        if not token:
+            st.error("请先填写 Bearer Token。")
+        else:
+            st.session_state.new_api_bearer_token = token
+            st.session_state.new_api_query_date = query_date
+            date_str = query_date.strftime("%Y-%m-%d")
+            with st.spinner(f"正在调用接口拉取 {date_str} 的场次..."):
+                try:
+                    records = fetch_all_movieshow(token, date_str, size=page_size)
+                except requests.HTTPError as exc:
+                    st.error(f"HTTP 错误：{exc.response.status_code} {exc.response.reason}")
+                    records = []
+                except Exception as exc:  # noqa: BLE001
+                    st.error(f"调用接口失败：{exc}")
+                    records = []
+
+            if records:
+                st.session_state.new_api_records = records
+                st.session_state.new_api_data_source = f"API 拉取（{len(records)} 条）"
+                source_df = records_to_source_df(records)
+                report_df, display_date = build_daily_report_from_source_df(
+                    source_df,
+                    selected_date_str=date_str,
+                )
+                st.session_state.new_api_report_df = report_df
+                st.session_state.new_api_display_date = display_date or query_date
+                if not report_df.empty:
+                    st.toast(f"成功生成 {len(report_df)} 条报表数据。", icon="✅")
+
+
+with input_tab_paste:
+    st.markdown("将接口返回的整个 JSON 文本（或 `data.records` 数组）粘贴到下方：")
+    paste_text = st.text_area(
+        "JSON 文本",
+        value=st.session_state.new_api_paste_text,
+        height=260,
+        placeholder='{"code":0,"data":{"records":[ ... ]}}',
+        key="new_api_paste_textarea",
+    )
+    fallback_date = st.date_input(
+        "用于补充缺失日期的回退值",
+        value=st.session_state.new_api_query_date,
+        key="new_api_paste_fallback_date",
+    )
+
+    if st.button("📥 解析并生成报表", key="new_api_parse_btn"):
+        st.session_state.new_api_paste_text = paste_text
+        try:
+            records = parse_pasted_response(paste_text)
+        except ValueError as exc:
+            st.error(str(exc))
+            records = []
+
+        if records:
+            st.session_state.new_api_records = records
+            st.session_state.new_api_data_source = f"粘贴 JSON（{len(records)} 条）"
+            source_df = records_to_source_df(records)
+            fallback_str = fallback_date.strftime("%Y-%m-%d")
+            report_df, display_date = build_daily_report_from_source_df(
+                source_df,
+                selected_date_str=fallback_str,
+            )
+            st.session_state.new_api_report_df = report_df
+            st.session_state.new_api_display_date = display_date or fallback_date
+            if not report_df.empty:
+                st.toast(f"成功生成 {len(report_df)} 条报表数据。", icon="✅")
+
+
+# ---------------------------------------------------------------------------
+# 报表展示与下载
+# ---------------------------------------------------------------------------
+st.divider()
+
+report_df: pd.DataFrame = st.session_state.new_api_report_df
+display_date = st.session_state.new_api_display_date
+
+if isinstance(report_df, pd.DataFrame) and not report_df.empty:
+    if display_date is None:
+        display_date = dt_date.today()
+    st.caption(f"当前数据来源：{st.session_state.new_api_data_source}")
+    st.markdown(f"#### {display_date.strftime('%Y-%m-%d')} 影片映出日累计报表")
+
+    st.dataframe(
+        report_df.style.format(
+            {"人数合计": "{:,.0f}", "座位数": "{:,.0f}", "上座率%": "{:.2f}%"}
+        ),
+        width="stretch",
+        hide_index=True,
+    )
+
+    total_attendance = pd.to_numeric(report_df.get("人数合计", 0), errors="coerce").fillna(0).sum()
+    st.markdown(
+        f"""
+        <div style="margin: 12px 0 18px; padding: 14px 18px; border-left: 6px solid #D83B01; background: #FFF4ED;">
+            <span style="font-size: 18px; font-weight: 700; color: #5C1F00;">已售人次：</span>
+            <span style="font-size: 26px; font-weight: 800; color: #D83B01;">{total_attendance:,.0f}</span>
+            <span style="font-size: 18px; font-weight: 700; color: #5C1F00;"> 人</span>
+        </div>
+        """,
+        unsafe_allow_html=True,
+    )
+
+    output_buffer = io.BytesIO()
+    report_df.to_excel(output_buffer, index=False, engine="openpyxl")
+    st.download_button(
+        label="📥 下载 XLSX 报表文件",
+        data=output_buffer.getvalue(),
+        file_name=f"{display_date.strftime('%Y-%m-%d')}_影片映出日累计报表.xlsx",
+        mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    )
+
+    with st.expander("查看原始接口数据（最多 10 条）"):
+        sample = st.session_state.new_api_records[:10]
+        st.json(sample)
+else:
+    st.info("尚无数据。请通过「API 拉取」或「粘贴 JSON」生成报表。")