某星球预约抢票脚本

代码如下：

import random
import re
import asyncio
import datetime
import time
import requests


def remove_duplicates(dict_list):
    seen = set()
    new_list = []
    for d in dict_list:
        frozen_dict = frozenset(d.items())
        if frozen_dict not in seen:
            seen.add(frozen_dict)
            new_list.append(d)
    return new_list


class PxqShowMonitor:
    def __init__(self):
        self.http = requests.Session()
        ua = ('Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) '
              'Version/16.6 Mobile/15E148 Safari/604.1')
        self.http.headers.update({
            'User-Agent': ua,
            'Src': 'H5',
            'Terminal-Src': 'H5'
        })
        self.api_src = 'H5'
        self.api_ver = "4.3.5-20240408181716"

    def get_city_list(self):
        """
        :return:
        """
        try:
            url = 'https://m.xxxx.com/cyy_gatewayapi/home/pub/v3/citys'
            params = {
                'src': self.api_src,
                'ver': self.api_ver
            }
            response = self.http.get(url, params=params)
            data = response.json()
            city_list = []
            for item in data.get('data').get('allCities'):
                city_list.append({
                    'city_id': item['cities'][0]['cityId'],
                    'city_name': item['cities'][0]['cityName'],
                    'site_id': item['cities'][0]['siteId'],
                })
            return city_list
        except Exception as e:
            print(f"获取城市列表失败, {e.args}")
            return []

    def get_show_list(self, city_id):
        """
        获取即将开抢的演唱会列表
        :return:
        """
        try:
            url = 'https://m.xxxxx.com/cyy_gatewayapi/home/pub/v3/floors'
            params = {
                'src': self.api_src,
                'ver': self.api_ver,
                'cityId': city_id
            }
            response = self.http.get(url, params=params)
            data = response.json()
            show_list = []
            for item in data['data']:
                for show in item['rooms'][0]['items']:
                    if 'latestSaleTime' in show:
                        show_list.append(show)
            return show_list
        except Exception as e:
            print(f'获取演出失败:{e.args}')

    def run(self):
        city_list = self.get_city_list()
        if not city_list:
            print("获取城市列表失败, 退出程序...")
            return

        all_show_list = []
        for city in city_list:
            print('正在获取{}即将开抢的演唱会列表...'.format(city['city_name']))
            show_list = self.get_show_list(city['city_id'])
            print('{},获取到{}个演唱会'.format(city['city_name'], len(show_list)))
            all_show_list.extend(show_list)
            time.sleep(2)

        show_list = []
        for item in all_show_list:
            # 计算三天后的日期时间对象
            three_days_later = datetime.datetime.now() + datetime.timedelta(days=3)

            # 将三天后的日期时间对象转换为毫秒时间戳
            three_days_later_timestamp = three_days_later.timestamp() * 1000

            if item.get('saleTime', None) and item['saleTime'] > three_days_later_timestamp:
                continue

            if item.get('saleTime', None):
                sale_time = datetime.datetime.fromtimestamp(item['saleTime'] / 1000).strftime('%Y-%m-%d %H:%M:%S')
            else:
                sale_time = item['latestSaleTime']
            show_list.append({
                'city_name': item['cityName'],
                'show_name': item['title'],
                'sale_time': sale_time,
                'id': item['id']
            })
        unique_dict_list = [dict(t) for t in {tuple(d.items()) for d in show_list}]
        msg = '### 近期开售的演出列表: \n'

        for item in unique_dict_list:
            msg += '- {}(id:{}), 开售时间:{}\n'.format(item['show_name'], item['id'], item['sale_time'])
        return msg


def server_jiang_notify(key, title, msg=''):
    """
        :param key:
        :param title:
        :param msg:
        :return:
        """
    try:
        url = f'https://sctapi.ftqq.com/{key}.send'
        data = {
            'title': title,
            'desp': msg,
        }
        response = requests.post(url, json=data)
        if response.json().get('code', -1) == 0:
            return True
        else:
            print("发送通知失败, {}".format(response.json()['message']))
    except Exception as e:
        print(f"发送通知失败:{e.args}")
    return False


def main():
    app = PxqShowMonitor()
    message = app.run()
    with open('README.md', 'r') as f:
        content = f.read()

    content = re.sub('<!-- SHOW_START -->([\s\S]*?)<!-- SHOW_END -->',
                     f'<!-- SHOW_START -->\n{message}\n<!-- SHOW_END -->', content)

    with open('README.md', 'w') as f:
        f.write(content)


if __name__ == '__main__':
    main()

说明

本文的Python脚本是一个自动化工具，用于监控即将开售的演出票务信息，并更新一个Markdown文件(`README.md`)来显示即将开售的演出列表。它利用了网络请求来获取信息，并对数据进行了处理和格式化，最后通过特定的API发送通知。让我们逐步分析这段代码的主要功能和组成部分。

类和函数

PxqShowMonitor类

– 初始化：设置HTTP会话头部，包括用户代理和API版本。

– get_city_list方法：获取一个城市列表，每个城市具有ID和名称，从预设的URL获取数据。

– get_show_list 方法：根据城市ID获取该城市即将开售的演出列表，筛选出包含`latestSaleTime`的演出。

– run方法：主要的执行流程，获取所有城市的演出列表，筛选出三天内即将开售的演出，并去重，然后格式化成消息字符串。

辅助函数

– remove_duplicate：从列表中移除重复的字典项。

– server_jiang_notify：发送通知到Server酱，一个常用于微信消息推送的服务。

– main` 函数：实例化`PxqShowMonitor`，运行并获取消息，然后将消息插入到`README.md`的指定部分。

工作流程

1. 初始化：创建`PxqShowMonitor`实例，配置必要的HTTP头部信息。

2. 获取数据：首先获取城市列表，然后遍历每个城市获取即将开售的演出信息。

3. 数据处理：对获取的演出数据进行过滤，只保留三天内开始售票的演出，并进行去重处理。

4. 消息格式化：将筛选后的演出信息格式化为Markdown文本。

5. 更新文件：使用正则表达式更新`README.md`文件中的特定部分。

6. 发送通知：可选地通过Server酱发送通知。

应用场景

这个脚本可以用于自动化监控票务网站上即将开售的演出信息，为用户提供及时的更新，非常适合那些需要第一时间抢购演出票的用户。此外，它还可以被集成到更大的系统中，提供定时任务运行，自动更新信息和通知用户。

总结

此脚本是一个自动化的工具，通过集成多个服务和处理步骤来实现一个完整的业务流程：从获取数据到处理、展示和通知。它展示了Python在数据抓取、处理和自动化方面的强大能力，尤其是在处理网络数据和自动化文件操作方面。

注意：

本文部分变量已做脱敏处理，仅用于测试和学习研究，禁止用于商业用途，不能保证其合法性，准确性，完整性和有效性，请根据情况自行判断。技术层面需要提供帮助，可以通过打赏的方式进行探讨。