GK/3.31GK/frame_code/clean_gk_flights_datas.py

import re
import time
import queue
import threading
from datetime import datetime

from bson.objectid import ObjectId
from my_logger import Logger
from flights_mongodb import mongo_con_parse
from flights_utils import time_format_conversion, get_now_time, put_data_by_queue, get_data_by_queue
from settings import CRAWL_DATE_CONF_STATUS_TAB
from yz_clean_utils import delete_mongodb_clean_datas, save_mongodb_clean_datas, \
    get_city_code, format_pc_kg_data
import json

lock = threading.Lock()


def convert_time_format(input_str):
    """"2025年5月30日 (週五) 下午3:35 => 20250530153500 """
    # 移除括號內的星期信息（如 "週五"）
    cleaned_str = re.sub(r'\s*\(.*?\)', '', input_str)

    # 將中文上午/下午轉換為 AM/PM 標記
    cleaned_str = cleaned_str.replace("上午", "AM").replace("下午", "PM")

    # 解析日期時間
    dt = datetime.strptime(cleaned_str, "%Y年%m月%d日 %p%I:%M")

    # 格式化為目標格式，並補齊秒數 (00)
    return dt.strftime("%Y%m%d%H%M%S")


class CleanGKDatas():

    def __init__(self):
        self.website = "gk"
        self.logger = Logger('./logs/{}_flights_clear.log'.format(self.website), 'debug').logger
        self.db = mongo_con_parse()
        self.info_tab = "flights_{}_info_tab".format(self.website)
        self.clean_info_tab = "clean_{}".format(self.info_tab)
        self.thread_numbers = 1
        self.task_queue = queue.Queue()
        self.count = 0

    def processing_data_one_search(self, find_info_data, thread_number=0, verify_flag=False):
        # global has_baggage, baggage
        from_airport_code = ''
        to_airport_code = ''
        search_date = ''
        end_flight_datas = []
        try:
            # print(find_info_data)
            # 每一次搜索
            from_city_code = find_info_data.get("search_from_city_code")
            to_city_code = find_info_data.get("search_to_city_code")

            crawl_date = find_info_data.get("crawl_date")
            search_date = find_info_data.get("search_date")
            # 转为保存数据库的出发日期（in_strftime_str：传入时间的格式） 注：为时间
            search_dep_time_base = time_format_conversion(search_date, in_strftime_str="%Y-%m-%d")

            # 提取网站源数据
            inner_data = find_info_data.get('inner_data')

            # 取gk航班
            flights = inner_data.get('Trips', [])[0].get("Flights", [])

            # 处理不同的航班
            for each_flight in flights:

                mid_flight_datas = []

                DisplayFlightInfo = each_flight.get("DisplayFlightInfo", {})  # 航班信息

                from_airport_code = DisplayFlightInfo.get('OriginAirport')  # 出发机场
                to_airport_code = DisplayFlightInfo.get('DestinationAirport')  # 到达机场代码

                """提取票价"""
                # 不同票价的余票 先设置为舱位的余票,因为每个票价没有对应的余票,
                # 所以先检查经济舱是否有余票,无余票就跳过 (商务舱不用管, 因为查询参数传入的是经济舱/ 而且网站响应商务舱的余票好像一直是 0)

                EconomyFlightInfo = DisplayFlightInfo.get('EconomyFlightInfo')  # 经济舱信息  里面是 余票数
                if EconomyFlightInfo['RemainingSeats'] == 0:
                    print('跳过(经济舱)余票数为 0 的航班')
                    continue
                # 余票
                seatCount = EconomyFlightInfo['RemainingSeats']

                Bundles = each_flight['Bundles']  # 票价列表
                # 税金(手续费)和基本价提取
                EconomyPriceBreakdown = json.loads(each_flight['EconomyPriceBreakdown'])  # 里面有税金, 是json字符串要转为对象
                price = EconomyPriceBreakdown['TotalFare']  # 基础票价
                adult_tax = EconomyPriceBreakdown['TotalCharges']  # 同一个航班是不变的

                # 提取每个票价
                for each_bundle in Bundles:
                    # 服务包代码
                    fareBasis = each_bundle['ServiceBundleCode']

                    #  只采集 最低不带行李的价格 和 一个最低带行李 的票价
                    # S000: 基本票價   P200: 基本加值套票 行李 + 座位 + 餐膳
                    if fareBasis not in ['S000', 'P200']:
                        continue

                    Amount = each_bundle['Amount']  # 增值服务费, 后面算总价要加上这个

                    # 托运行李
                    if fareBasis == 'S000':
                        pc_amount, kg_amount = '0', '0'  # 无托运行李
                        has_baggage, baggage = format_pc_kg_data(pc_amount, kg_amount)
                    if fareBasis == 'P200':
                        pc_amount, kg_amount = '1', '20'  # 1件行李20公斤
                        has_baggage, baggage = format_pc_kg_data(pc_amount, kg_amount)

                    """处理中转"""
                    segment_elements = []
                    Legs = DisplayFlightInfo.get('Legs')  # 中转航段信息, 最多有3个

                    # 可能是直达或转乘
                    for each_leg in Legs:
                        '''理论航班号信息'''
                        CarrierCode = each_leg['FlightDesignator']['CarrierCode']    # 航空公司
                        FlightNumber = each_leg['FlightDesignator']['FlightNumber']  # 航班号
                        flight_number = f"{CarrierCode}{FlightNumber}"  # 拼接航班号

                        '''实际航班号信息, gk没有这些, ??? 该如何??? => 略 '''
                        # operatingAirlineCode = each_leg['operatingAirlineCode']  # 营运航空公司代码
                        # operatingFlightno = each_leg['operatingFlightno']
                        # operating_flight_number = f"{operatingAirlineCode}{operatingFlightno}"

                        '''机型'''
                        # 空中巴士 A320-200 (代碼 32J，全經濟艙配置）
                        aircraftCode = each_leg['Equipment']['Type']

                        '''出发/到达时间'''
                        departDate = each_leg["DisplayStd"]  # 计划起飞时间  2025年5月31日 (週六) 上午2:15
                        arrivalDate = each_leg["DisplaySta"]  # 计划到达时间

                        dep_time = convert_time_format(departDate)  # 时间格式转换
                        arr_time = convert_time_format(arrivalDate)

                        '''出发/到达机场和城市'''
                        depart_airport = each_leg['DepartureStation']  # 出发机场
                        depart_city = get_city_code(self.db, self.website, depart_airport)  # 数据库查询对应城市代码

                        arrival_airport = each_leg['ArrivalStation']  # 到达机场
                        arrival_city = get_city_code(self.db, self.website, arrival_airport)

                        '''舱位类型'''
                        CabinType = each_leg['CabinType']

                        # 航站楼
                        # DisplayDepartureAirportTerminal = each_leg['DisplayDepartureAirportTerminal']  # 出发
                        # DisplayArrivalAirportTerminal = each_leg['DisplayArrivalAirportTerminal']      # 到达

                        segment_element_demo = {
                            "flight_number": flight_number,  # 航班号
                            "operating_flight_number": flight_number,  # 实际营运航班号 (gk没有这个和上面写为同一个)
                            "dep_air_port": depart_airport,  # 出发机场code str
                            "dep_city_code": depart_city,  # 出发城市code
                            "dep_time": dep_time,               # 出发时间 格式YYYYMMDDHHMM str
                            "arr_air_port": arrival_airport,  # 抵达机场code str
                            "arr_city_code": arrival_city,  # 抵达城市code
                            "arr_time": arr_time,  # 抵达时间 格式YYYYMMDDHHMM str
                            "cabin": CabinType,  # 舱位单字母
                            "carrier": CarrierCode,  # 航空公司代码
                            "aircraft_code": aircraftCode,  # 机型 str
                            "cabin_class": 1,  # 舱位等级, 因为捷星日本航空（GK）的航班目前均为全经济舱​​，故这里也可写死
                            "stop_cities": "",
                            "has_baggage": has_baggage,  # 是否有托运行李
                            "baggage": baggage,  # 行李配重 1-23 表示一件行李 23kg
                            "fareBasis": fareBasis,  # 在同一航班的不同舱位中是唯一值，主要用来区分同一航班的不同舱位，
                        }
                        segment_elements.append(segment_element_demo)

                    # 查询出发时间（如果响应没有，可能是入参传递的）
                    search_dep_time = segment_elements[0].get("dep_time") if segment_elements else search_dep_time_base

                    flight_datas = {
                        "from_city_code": from_city_code,
                        "search_dep_time": search_dep_time,
                        "to_city_code": to_city_code,
                        "currency": 'JPY',  # 网站默认返回搜索地区的国家货币价格, 采集的全都是 日本的航班 故可写死
                        "adult_price": int(Amount+price),  # 基本价格
                        "adult_tax": int(adult_tax),  # 税金
                        "adult_total_price": int(Amount+price+adult_tax),  # 再加 基本价和税金
                        "route": "",
                        "seats_remaining": seatCount,
                        "segments": segment_elements,
                        "source_website": self.website,
                        "crawl_date": crawl_date
                    }

                    # print(flight_datas)

                    if verify_flag:
                        flight_datas["verify_date"] = crawl_date  # 验价分支
                        verify_time = get_now_time()
                        flight_datas["verify_time"] = verify_time
                        # print(flight_datas)
                    mid_flight_datas.append(flight_datas)
                # 注意：这里已经跳出bundle循环，将当前航班的所有票价记录（mid_flight_datas）扩展到总结果
                end_flight_datas.extend(mid_flight_datas)

            if verify_flag:
                # 验价分支，
                search_from_city_code = find_info_data.get("search_from_city_code")
                search_to_city_code = find_info_data.get("search_to_city_code")
                search_date = find_info_data.get("search_date")
                end_search_date = time_format_conversion(
                    search_date,
                    in_strftime_str="%Y-%m-%d",
                    out_strftime_str="%Y%m%d"
                )

                delete_many_filter = {
                    "from_city_code": search_from_city_code,
                    "to_city_code": search_to_city_code,
                    "search_dep_time": {"$regex": r"^{}".format(end_search_date)}
                }
                delete_mongodb_clean_datas(self.db, self.clean_info_tab, self.logger, delete_many_filter)
                # 重新保存最新的采集数据
                if len(end_flight_datas) > 0:
                    save_mongodb_clean_datas(self.db, self.clean_info_tab, self.website, self.logger, end_flight_datas, thread_number, True)

        except Exception as e:
            self.logger.error(f"thread_number:{thread_number} clean error: {from_airport_code}---{to_airport_code}---{search_date}---{str(e)}")
        finally:
            return end_flight_datas

    def processing_data(self, thread_number):
        while 1:
            log_ob_ids = []
            try:
                ids = get_data_by_queue(self.task_queue)
                ob_ids = [ObjectId(i) for i in ids]
                log_ob_ids = ob_ids
                # 批量查询 _id 包含在给定列表 ob_ids 中的文档
                find_info_datas = self.db.get_collection(self.info_tab).find(
                    {"_id": {"$in": ob_ids}}
                )
                final_flight_datas = []
                # 每一次搜索航线
                for find_info_data in find_info_datas:
                    with lock:
                        self.count += 1
                    if self.count % 50 == 0:
                        self.logger.info("thread_number:{0}, clean count: {1}".format(
                            thread_number, self.count))
                    # print(find_info_data)
                    # exit()
                    end_flight_datas = self.processing_data_one_search(find_info_data, thread_number)
                    final_flight_datas.extend(end_flight_datas)

                # 更改这些info表里的数据清理状态为1
                update_result = self.db.get_collection(self.info_tab).update_many(
                    {"_id": {"$in": ob_ids}},
                    {"$set": {"clean_status": 1}}
                )
                self.logger.info(f"Updated documents: {update_result.modified_count}")
                if len(final_flight_datas) > 0:
                    save_mongodb_clean_datas(self.db, self.clean_info_tab, self.website, self.logger, final_flight_datas, thread_number, True)

            except Exception as e:
                self.logger.error(f"thread_number:{thread_number}, log_ob_ids:{log_ob_ids}, clean unknown err:{str(e)}")
            finally:
                self.task_queue.task_done()

    def run_threading(self):
        for thread_number in range(1, self.thread_numbers+1):
            t = threading.Thread(
                target=self.processing_data,
                args=(thread_number,)
            )
            t.daemon = True
            t.start()
        self.task_queue.join()

    def split_datas(self, datas, num):
        return [datas[i:i + num] for i in range(0, len(datas), num)]

    def run(self):
        # 每次清除之前清洗的数据
        # 查询已经采集完成但未清洗的数据批次
        while 1:
            # 取最新采集的数据
            find_crawl_conf_datas = self.db.get_collection(
                CRAWL_DATE_CONF_STATUS_TAB).find(  # find查询
                {
                    "website": self.website,
                    # "crawl_status": 1,  # 反复测试清洗逻辑
                    "clean_status": 0
                },  # 查询条件
                sort=[('_id', -1)]  # 排序规则，按 _id 降序
            ).limit(1)  # 仅返回第一条记录（即最新的一条文档）

            for find_crawl_conf_data in find_crawl_conf_datas:
                self.count = 0
                self.logger.info("start clean crawl_date: {}".format(
                    find_crawl_conf_data.get("crawl_date")
                ))
                find_info_datas = self.db.get_collection(self.info_tab).find(
                    {
                        "crawl_date": find_crawl_conf_data.get("crawl_date"),
                        "website": self.website,
                        # "clean_status": 1,  # 反复测试清洗逻辑
                        "clean_status": 0
                    }
                )
                find_id_datas = [str(i.get("_id")) for i in find_info_datas]
                find_id_datas_splits = self.split_datas(find_id_datas, 5)
                # 一个队列里最多放5份info数据
                for find_id_data_split in find_id_datas_splits:
                    put_data_by_queue(self.task_queue, find_id_data_split)

                self.run_threading()  # 多线程清理
                self.logger.info("batch clean all counts: {}".format(self.count))

                # 判断整体结束的条件
                if find_crawl_conf_data.get("crawl_status") == 1:
                    self.update_clean_date_status(find_crawl_conf_data)
                    self.logger.info("end clean crawl_date: {}".format(
                        find_crawl_conf_data.get("crawl_date")
                    ))
                    # 整体结束后, 统一计算本次清理的记录数, 为零则告警
                    clean_crawl_date_counts = self.db.get_collection(
                        self.clean_info_tab).count_documents(
                        {
                            "crawl_date": find_crawl_conf_data.get("crawl_date")
                        }
                    )
                    self.logger.info("crawl_date: {0}, insert counts: {1}".format(
                        find_crawl_conf_data.get("crawl_date"), clean_crawl_date_counts
                    ))

                    # 发送钉钉通知
                    try:
                        from clean_datas_send_notice import CheckCrawlDatas
                        C = CheckCrawlDatas()
                        C.send_website_spiders_crawl_date_counts(
                            self.website, find_crawl_conf_data.get("crawl_date"),
                            clean_crawl_date_counts
                        )
                    except Exception as e:
                        self.logger.info("send dingding error: {0}".format(str(e)))

            time.sleep(10 * 1)

    def update_clean_date_status(self, find_crawl_conf_data):
        # 更新 采集批次时间 状态
        self.db.get_collection(CRAWL_DATE_CONF_STATUS_TAB).update_one(
             {
                 "_id": ObjectId(find_crawl_conf_data.get("_id"))
             },
             {
                 "$set":
                     {
                         "clean_status": 1
                     }
             }
        )


if __name__ == "__main__":
    C = CleanGKDatas()
    C.run()