基于ModelArts的电影实时票房数据分析与预测

logic-crafter

405人浏览 · 2025-09-17 23:23:52

logic-crafter · 2025-09-17 23:23:52 发布

本项目大数据采集和线性回归的基础原理，使用ModelArts进行开发工作，编写Python代码从票房网站爬取数据到本地CSV文件，然后针对CSV文件的数据进行转换，分析，建模，最后预测未来的票房数据。以下通过猫眼《捕风追影》电影示例

一、创建Notebook

二、点击“File\New Launcher”进入Launcher页面后，点击Terminal图标进入Terminal页面

在Terminal终端，输入下面的命令安装schedule服务包，如下图，安装成功后会显示成功的信息：

                                pip install schedule

三、爬取票房数据

点击菜单File\New Launcher，选择PyTorch-XX, 打开一个代码编辑窗口。

下面的代码将爬取票房网站的数据，经过解码和转换后，到本地CSV文件，程序开启了一个定时任务，它每分钟执行一次，意味着每一份钟爬取票房数据，然后往CSV文件写入一行票房数据。

import requests
import schedule
import time
import csv
import re
import os
from datetime import datetime
import html
from fontTools.ttLib import TTFont
import json
import base64


# 发送HTTP请求获取实时票房数据
def fetch_box_office_data():
    # 网址为存放动态数据的网址:猫眼电影API的URL
    MOVIE_WEBSITE_URL = "https://piaofang.maoyan.com/dashboard-ajax/movie"
    HEADERS = {
        "Accept": "application/json, text/plain, */*",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        'Referer': 'https://piaofang.maoyan.com/dashboard/movie',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36 Edg/134.0.0.0'
    }
    
    # 使用当前时间戳
    current_timestamp = int(time.time() * 1000)
    
    # 基本参数
    params = {
        "orderType": "0",
        "uuid": f"auto-gen-{current_timestamp}",
        "timeStamp": str(current_timestamp),
        "User-Agent": "TW96aWxsYS81LjAgKFdpbmRvd3MgTlQgMTAuMDsgV2luNjQ7IHg2NCkgQXBwbGVXZWJLaXQvNTM3LjM2IChLSHRNTCwgbGlrZSBHZWNrbykgQ2hyb21lLzEzNC4wLjAuMCBTYWZhcmkvNTM3LjM2IEVkZy8xMzQuMC4wLjA%3D",
        "index": "84",
        "channelId": "40009",
        "sVersion": "2",
        "signKey": f"sign-{current_timestamp}",
        "WukongReady": "h5"
    }

    try:
        x = requests.get(MOVIE_WEBSITE_URL, headers=HEADERS, params=params, timeout=10)
        if x.status_code == 200:
            json_data = x.json()
            return json_data
        else:
            print(f"Failed to fetch data, status code: {x.status_code}")
            print(f"Response: {x.text[:500]}...")
            return None
    except Exception as e:
        print(f"Error fetching data: {e}")
        return None


def writedata(jdata, writer):
    '''提取字体链接'''
    if not jdata or 'fontStyle' not in jdata:
        print("No fontStyle found in response")
        return
        
    fontStyle = jdata['fontStyle']
    
    # 直接从fontStyle中提取字体URL
    pattern = r'url\("([^"]+\.woff)"\)'
    match = re.search(pattern, fontStyle)
    
    if not match:
        print("Failed to extract font URL with pattern")
        return
    
    font_url = match.group(1)
    if font_url.startswith('//'):
        font_url = 'https:' + font_url
    elif not font_url.startswith('http'):
        font_url = 'https://' + font_url
        
    print("Extracted font_url:", font_url)

    # 获取当前时间
    current_time = datetime.now()
    formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
    
    if "movieList" not in jdata or "list" not in jdata["movieList"]:
        print("No movie list found in response")
        return
        
    print(f"Found {len(jdata['movieList']['list'])} movies in response")
    
    found_target_movie = False
    for movie in jdata["movieList"]["list"]:
        if "movieInfo" not in movie:
            continue
            
        movie_info = movie["movieInfo"]
        if "movieId" not in movie_info:
            continue
            
        movie_id = movie_info["movieId"]
        movie_name = movie_info.get("movieName", "Unknown")
        print(f"Processing movie: ID={movie_id}, Name={movie_name}")
            
        # 只处理《捕风追影》 (根据电影名称匹配)
        if "捕风追影" in movie_name:
            found_target_movie = True
            movie_data = {
                "movieId": movie_id,
                "movieName": movie_name,
                "dataTimestamp": formatted_time
            }
            
            # 处理综合票房
            if "boxSplitUnit" in movie and "num" in movie["boxSplitUnit"]:
                encoded_box_office = movie["boxSplitUnit"]["num"]
                movie_data["boxSplitUnitNum"] = decode_data(font_url, encoded_box_office)
                print(f"Box office: {encoded_box_office} -> {movie_data['boxSplitUnitNum']}")
            else:
                movie_data["boxSplitUnitNum"] = "N/A"
                print("No boxSplitUnit found")
                
            # 处理分账票房
            if "splitBoxSplitUnit" in movie and "num" in movie["splitBoxSplitUnit"]:
                encoded_split_box_office = movie["splitBoxSplitUnit"]["num"]
                movie_data["splitBoxSplitNum"] = decode_data(font_url, encoded_split_box_office)
                print(f"Split box office: {encoded_split_box_office} -> {movie_data['splitBoxSplitNum']}")
            else:
                movie_data["splitBoxSplitNum"] = "N/A"
                print("No splitBoxSplitUnit found")
                
            # 写入其他字段
            fields = ["avgSeatView", "avgShowView", "boxRate", "showCount", 
                     "showCountRate", "splitBoxRate", "sumBoxDesc", "sumSplitBoxDesc"]
            
            for field in fields:
                movie_data[field] = movie.get(field, "N/A")
            
            # 写入数据行
            writer.writerow(movie_data.values())
            print(f"Data for {movie_name} saved at {formatted_time}")
            break  # 找到目标电影后跳出循环
            
    if not found_target_movie:
        print("Target movie (捕风追影) not found in response")


# 将数据保存到CSV文件
def save_data_to_csv(json_data, filename="box_office.csv"):
    if not json_data:
        print("No data to save")
        return
        
    header = ["avgSeatView", "avgShowView", "boxRate", "showCount", "showCountRate", 
              "splitBoxRate", "sumBoxDesc", "sumSplitBoxDesc", "movieId", 
              "movieName", "boxSplitUnitNum", "splitBoxSplitNum", "dataTimestamp"]

    # 检查文件是否存在
    file_exists = os.path.exists(filename)
    
    with open(filename, 'a' if file_exists else 'w', encoding='utf-8', newline='') as csvfile:
        writer = csv.writer(csvfile)
        
        # 如果文件不存在，写入列名
        if not file_exists:
            writer.writerow(header)
            print(f"文件 '{filename}' 不存在，创建文件并写入列名...")
        else:
            print(f"文件 '{filename}' 存在, 写入新数据。")
            
        # 写入数据
        writedata(json_data, writer)
        
    print(f"Data saved to {filename}")


def decode_data(font_url, encoded_text):
    """
    破解加密字体函数
    参数:
        font_url (str): 字体文件的URL
        encoded_text (str): 加密的文本
    返回:
        str: 解密后的文本
    """
    if not encoded_text or encoded_text == "N/A":
        return "N/A"
    
    # 猫眼字体映射表（可能需要根据实际情况调整）
    font_mapping = {
        '&#xeaf6;': '0', '&#xf23f;': '0', '&#xeba2;': '0', '&#xea6f;': '0',
        '&#xf85e;': '1', '&#xe6d5;': '1', '&#xedba;': '1', '&#xed8f;': '1',
        '&#xe99c;': '2', '&#xec68;': '2', '&#xf0f0;': '2', '&#xe886;': '2',
        '&#xe8d7;': '3', '&#xf615;': '3', '&#xe85f;': '3', '&#xe583;': '3',
        '&#xe8ee;': '4', '&#xeab3;': '4', '&#xefe9;': '4', '&#xf16b;': '4',
        '&#xecdc;': '5', '&#xef74;': '5', '&#xed4f;': '5', '&#xe83d;': '5',
        '&#xf7ff;': '6', '&#xe5ac;': '6', '&#xf70e;': '6', '&#xf23f;': '6',
        '&#xeb92;': '7', '&#xe317;': '7', '&#xe916;': '7', '&#xe132;': '7',
        '&#xf1fc;': '8', '&#xe1b7;': '8', '&#xe83f;': '8', '&#xf05a;': '8',
        '&#xe9ea;': '9', '&#xe274;': '9', '&#xed98;': '9', '&#xec4b;': '9',
    }
    
    # 直接使用映射表替换HTML实体
    decoded_text = encoded_text
    for entity, digit in font_mapping.items():
        decoded_text = decoded_text.replace(entity, digit)
    
    # 移除可能残留的HTML实体
    decoded_text = re.sub(r'&#x[0-9a-f]+;', '', decoded_text)
    
    return decoded_text


# 主函数
def main():
    data = fetch_box_office_data()
    if data:
        save_data_to_csv(data)
    else:
        print("No data to save")


def job():
    print("Fetching and saving data...")
    main()


# 手动调用一次任务函数，确保任务立即启动
job()

# 间隔运行。
schedule.every(1).minutes.do(job)

# 启动定时任务
try:
    while True:
        schedule.run_pending()
        time.sleep(1)
except KeyboardInterrupt:
    print("程序已停止")

数据将保存到文件（box_office.csv）

最前面爬取的数据字体文件解析存在问题

最终按上面修改后的代码爬取的数据正常，从22行开始。

四、读取数据并分析数据

点击菜单File\New Launcher，选择PyTorch-XX, 打开一个代码编辑窗口，在单元格中输入以下代码，结果如下，因为我的数据从22行开始才正确，所以读取数据从22行开始。

#-*-coding:utf-8-*
import sklearn.model selection as msimport sklearn.linear model as lm
import sklearn.metrics as meansquarederror
import matplotlib.pyplot as plt
from datetime import datetime
import pandas as pd
#跳过前21行错误数据，从第22行开始读取
datas = pd.read csv("./box office.csv",skiprows=22, header=None)
datas .info()

新增一代码行，在单元格中输入以下代码，获得前5行数据信息,如下图所示

datas .head()

对字段进行相应的数据处理，在单元格中输入以下代码，如下图所示：

# 首先为数据框指定列名
column_names = [
    'filmId', 'filmName', 'dataTimestamp', 'avgSeatView', 'avgShowView', 
    'boxRate', 'showCount', 'showCountRate', 'boxSplitUnitNum', 
    'splitBoxRate', 'splitBoxSplitNum', 'boxOffice', 'splitBoxOffice'
]

# 为数据框指定列名
datas.columns = column_names

# 处理缺失值 - 这里改为 inplace=True 以实际删除缺失值
datas.dropna(inplace=True)

# 先处理无效值，将 '.' 替换为 NaN
datas['avgSeatView'] = datas['avgSeatView'].replace('.', pd.NA)
datas['boxRate'] = datas['boxRate'].replace('.', pd.NA)
datas['showCountRate'] = datas['showCountRate'].replace('.', pd.NA)
datas['splitBoxRate'] = datas['splitBoxRate'].replace('.', pd.NA)

# 再次删除包含无效值的行
datas.dropna(inplace=True)

# 处理%值 - 添加错误处理
def safe_convert_percent(x):
    try:
        # 确保x是字符串并去除%号
        x_str = str(x).strip()
        if x_str.endswith('%'):
            return float(x_str.strip('%')) / 100
        else:
            return float(x_str)
    except:
        return pd.NA

# 应用安全的转换函数
datas['avgSeatView'] = datas['avgSeatView'].apply(safe_convert_percent)
datas['boxRate'] = datas['boxRate'].apply(safe_convert_percent)
datas['showCountRate'] = datas['showCountRate'].apply(safe_convert_percent)
datas['splitBoxRate'] = datas['splitBoxRate'].apply(safe_convert_percent)

# 处理时间值
def safe_convert_timestamp(x):
    try:
        return datetime.strptime(x, "%Y-%m-%d %H:%M:%S").timestamp()
    except:
        return pd.NA

datas['dataTimestamp'] = datas['dataTimestamp'].apply(safe_convert_timestamp)

# 删除任何转换失败的行
datas.dropna(inplace=True)

# 检查需要编码的列的唯一值
columns_to_encode = ["avgSeatView", "avgShowView", "boxRate", "showCount", 
                    "showCountRate", "splitBoxRate", "boxSplitUnitNum", "splitBoxSplitNum"]

for column in columns_to_encode:
    print(f"\n{column}:")
    print(pd.unique(datas[column]))

结果如下：

五、数据训练和预测

1、拆分数据集。选取特征数据列和目标数据列，将特征数据（排片场次、排片占比、时间）拆分为训练集和测试集，然后对现有数据进行训练和预测，并打印均方误差，实际值与预测值的对比图，在单元格中输入以下代码：

features = ['showCount','showCountRate','dataTimestamp']
target = 'boxSplitUnitNum'

X = datas[features]
y = datas[target]

##划分训练集和测试集
Xtrain, Xtest, ytrain, ytest = ms.train_test_split(X, y, test_size=0.2, random_state=42)

##训练模型
model = lm.LinearRegression()
model.fit(Xtrain, ytrain)

#预测
ypred = model.predict(Xtest)

##评估模型
mse = meansquarederror.mean_squared_error(ytest, ypred)
print(f'均方误差:{mse}')

##绘制实际值与预测值对比图
plt.figure(figsize=(10, 6))
plt.plot(ytest.values, label='Actual')
plt.plot(ypred, label='Predicted')
plt.xlabel('Time')
plt.ylabel('Box Office')
plt.title('Box Office Prediction vs Actual')
plt.legend()
plt.show()

2、定义模型并预测输出。基于模型进行预测，修改正确的showCount，showCountRate，future_time， showCount，showCountRate的值可以直接从生成的文件（nezha2_box_office.csv）中拷贝， future_time可以是现在时间加上几分钟到几小时不等，在单元格中输入以下代码：

##avgSeatView：上座率， avgShowView：场均人次， boxRate：票房占比， showCount：排片场次，showCountRate：排片占比，
##预测未来票房
future_time = '2025-09-18 6:00:00'
futuredata = pd.DataFrame({
    'showCount': ['117665'],
    'showCountRate': [0.329],
    'dataTimestamp': [datetime.strptime(future_time, "%Y-%m-%d %H:%M:%S").timestamp()]
})

futurepredictions = model.predict(futuredata)
print(f'未来票房预测， 预测时间点:{future_time}, 预测的综合票房： {futurepredictions}')

腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

腾讯云开发者社区

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git