提交预测的后续处理，用于对接票务规则

follow_up.py

@@ -0,0 +1,221 @@
+import os
+import datetime
+import pandas as pd
+from config import mongodb_config
+
+
+def follow_up_handle():
+    '''后续处理'''
+    object_dir = "./predictions_0"
+
+    # 检查目录是否存在
+    if not os.path.exists(object_dir):
+        print(f"目录不存在: {object_dir}")
+        return
+    
+    # 获取所有以 future_predictions_ 开头的 CSV 文件
+    csv_files = []
+    for file in os.listdir(object_dir):
+        if file.startswith("future_predictions_") and file.endswith(".csv"):
+            csv_files.append(file)
+    
+    if not csv_files:
+        print(f"在 {object_dir} 中没有找到 future_predictions_ 开头的 CSV 文件")
+        return
+    
+    csv_files.sort()
+    
+    # 调试分支
+    # target_time = '202602251300'
+    # matching_files = [f for f in csv_files if target_time in f]
+    # if matching_files:
+    #     last_csv_file = matching_files[0]
+    #     print(f"指定时间的文件: {last_csv_file}")
+    # else:
+    #     print(f"未找到时间 {target_time} 的预测文件")
+
+    # 正式分支
+    last_csv_file = csv_files[-1]   # 只看最新预测的文件
+    print(f"最新预测文件: {last_csv_file}")
+
+    # 读取最新预测文件
+    last_csv_path = os.path.join(object_dir, last_csv_file)
+    df_last_predict = pd.read_csv(last_csv_path)
+
+    df_last_predict_will_drop = df_last_predict[df_last_predict["will_price_drop"] == 1].reset_index(drop=True)
+    print(f"最新预测文件中，预测降价的航班有 {len(df_last_predict_will_drop)} 条")
+
+    # 建一张 维护表 keep_info.csv
+    keep_info_path = os.path.join(object_dir, "keep_info.csv")
+    key_cols = ["city_pair", "flight_day", "flight_number_1", "flight_number_2"]
+
+    df_last_predict_will_drop = df_last_predict_will_drop.drop_duplicates(
+        subset=key_cols, keep="last"
+    ).reset_index(drop=True)
+
+    # 读取维护表
+    if os.path.exists(keep_info_path):
+        try:
+            df_keep_info = pd.read_csv(keep_info_path)
+        except Exception as e:
+            print(f"读取维护表失败: {keep_info_path}, error: {str(e)}")
+            df_keep_info = pd.DataFrame()
+    else:
+        df_keep_info = pd.DataFrame()
+
+    # 初始化维护表
+    if df_keep_info.empty:
+        df_keep_info = df_last_predict_will_drop.copy()
+        df_keep_info["keep_flag"] = 1
+        df_keep_info.to_csv(keep_info_path, index=False, encoding="utf-8-sig")
+        print(f"维护表已初始化: {keep_info_path} (rows={len(df_keep_info)})")
+    # 已存在维护表
+    else:
+        if "keep_flag" not in df_keep_info.columns:
+            df_keep_info["keep_flag"] = 0
+    
+        df_keep_info["keep_flag"] = (
+            pd.to_numeric(df_keep_info["keep_flag"], errors="coerce")
+            .fillna(0)
+            .astype(int)
+        )
+
+        missing_cols = [c for c in key_cols if c not in df_keep_info.columns]
+        if missing_cols:
+            print(f"维护表缺少字段: {missing_cols}, path={keep_info_path}")
+            return
+        
+        for c in key_cols:
+            df_last_predict_will_drop[c] = df_last_predict_will_drop[c].astype(str)
+            df_keep_info[c] = df_keep_info[c].astype(str)
+
+        df_keep_info = df_keep_info.drop_duplicates(subset=key_cols, keep="last").reset_index(drop=True)
+
+        # 提取两者的标志位
+        df_last_keys = df_last_predict_will_drop[key_cols].drop_duplicates().reset_index(drop=True)
+        df_keep_keys = df_keep_info[key_cols].drop_duplicates().reset_index(drop=True)
+
+        df_last_with_merge = df_last_predict_will_drop.merge(
+            df_keep_keys, on=key_cols, how="left", indicator=True
+        )
+        # 场景一: 如果某一行数据在 df_last_predict_will_drop 出现，没有在 df_keep_info 里
+        df_to_add = (
+            df_last_with_merge.loc[df_last_with_merge["_merge"] == "left_only"]
+            .drop(columns=["_merge"])
+            .copy()
+        )
+        # keep_flag 设为 1
+        if not df_to_add.empty:
+            df_to_add["keep_flag"] = 1
+        
+        df_keep_with_merge = df_keep_info.reset_index().merge(
+            df_last_keys, on=key_cols, how="left", indicator=True
+        )
+        # 场景二: 如果某一行数据在 df_last_predict_will_drop 和 df_keep_info 里都出现
+        matched_idx = df_keep_with_merge.loc[df_keep_with_merge["_merge"] == "both", "index"].tolist()
+        # 场景三: 如果某一行数据在 df_last_predict_will_drop 没有出现，却在 df_keep_info 里都出现
+        keep_only_idx = df_keep_with_merge.loc[df_keep_with_merge["_merge"] == "left_only", "index"].tolist()
+
+        # 符合场景二的索引 (在 df_keep_with_merge 中)
+        if matched_idx:
+            df_matched_keys = df_keep_info.loc[matched_idx, key_cols]
+            df_latest_matched = df_matched_keys.merge(
+                df_last_predict_will_drop, on=key_cols, how="left"
+            )
+            # 将 df_keep_info 的 df_matched_keys 的内容更新为 df_last_predict_will_drop 里对应的内容
+            update_cols = [c for c in df_last_predict_will_drop.columns if c not in key_cols]
+            for c in update_cols:
+                if c == "keep_flag":
+                    continue
+                if c not in df_keep_info.columns:
+                    df_keep_info[c] = pd.NA
+                df_keep_info.loc[matched_idx, c] = df_latest_matched[c].values
+
+            # 重新标记 原来是1 -> 0  原来是0 -> 0  原来是-1 -> 1
+            old_flags = df_keep_info.loc[matched_idx, "keep_flag"]
+            df_keep_info.loc[matched_idx, "keep_flag"] = old_flags.apply(
+                lambda x: 0 if x in (0, 1) else (1 if x == -1 else 1)
+            )
+        
+        # 符合场景三的索引 (在 df_keep_with_merge 中)
+        if keep_only_idx:
+            # 如果 df_keep_info 的 keep_flag 为-1，此时标记为-2
+            mask_keep_only = df_keep_info.index.isin(keep_only_idx)  # 布尔索引序列
+            mask_to_remove = mask_keep_only & (df_keep_info["keep_flag"] == -1)
+            df_keep_info.loc[mask_to_remove, "keep_flag"] = -2
+
+            # 如果 df_keep_info 的 keep_flag 大于等于0
+            mask_need_observe = mask_keep_only & (df_keep_info["keep_flag"] >= 0)  # 布尔索引序列
+            if mask_need_observe.any():
+                if "hours_until_departure" not in df_keep_info.columns:
+                    df_keep_info.loc[mask_need_observe, "keep_flag"] = -1
+                else:
+                    hud = pd.to_numeric(
+                        df_keep_info.loc[mask_need_observe, "hours_until_departure"],
+                        errors="coerce",
+                    )
+                    # hours_until_departure自动减1
+                    new_hud = hud - 1
+                    df_keep_info.loc[mask_need_observe, "hours_until_departure"] = new_hud
+
+                    idx_eq13 = mask_need_observe.copy()
+                    idx_eq13.loc[idx_eq13] = hud.eq(13)   # 原hours_until_departure等于13 
+
+                    idx_gt13 = mask_need_observe.copy()
+                    idx_gt13.loc[idx_gt13] = hud.gt(13)   # 原hours_until_departure大于13
+
+                    idx_other = mask_need_observe & ~(idx_eq13 | idx_gt13)  # 原hours_until_departure小于13
+
+                    idx_eq13_gt4 = idx_eq13 & new_hud.gt(4)
+                    idx_eq13_eq4 = idx_eq13 & new_hud.eq(4)
+                    idx_eq13_lt4 = idx_eq13 & new_hud.lt(4)
+
+                    df_keep_info.loc[idx_eq13_gt4, "keep_flag"] = 0
+                    df_keep_info.loc[idx_eq13_eq4, "keep_flag"] = -1
+                    df_keep_info.loc[idx_eq13_lt4, "keep_flag"] = -2
+
+                    df_keep_info.loc[idx_gt13, "keep_flag"] = -1
+
+                    idx_other_gt4 = idx_other & new_hud.gt(4)
+                    idx_other_eq4 = idx_other & new_hud.eq(4)
+                    idx_other_lt4 = idx_other & new_hud.lt(4)
+
+                    df_keep_info.loc[idx_other_gt4, "keep_flag"] = 0
+                    df_keep_info.loc[idx_other_eq4, "keep_flag"] = -1
+                    df_keep_info.loc[idx_other_lt4, "keep_flag"] = -2
+
+        # 将 df_to_add 添加到 df_keep_info 之后
+        add_rows = len(df_to_add) if "df_to_add" in locals() else 0
+        if add_rows:
+            df_keep_info = pd.concat([df_keep_info, df_to_add], ignore_index=True)
+        
+        # 移除 keep_flag 为 -2 的行
+        before_rm = len(df_keep_info)
+        df_keep_info = df_keep_info.loc[df_keep_info["keep_flag"] != -2].reset_index(drop=True)
+        rm_rows = before_rm - len(df_keep_info)
+
+        # 保存更新后的 df_keep_info 到csv文件
+        df_keep_info.to_csv(keep_info_path, index=False, encoding="utf-8-sig")
+        print(
+            f"维护表已更新: {keep_info_path} (rows={len(df_keep_info)} add={add_rows} rm={rm_rows})"
+        )
+
+    # ================================================================
+    # for idx, row in df_last_predict_will_drop.iterrows():
+    #     city_pair = row['city_pair']
+    #     flight_day = row['flight_day']
+    #     flight_number_1 = row['flight_number_1']
+    #     flight_number_2 = row['flight_number_2']
+    #     baggage = row['baggage']
+    #     from_city_code = city_pair.split('-')[0]
+    #     to_city_code = city_pair.split('-')[1]
+    #     from_day = datetime.datetime.strptime(flight_day, '%Y-%m-%d').strftime('%Y%m%d')
+    #     baggage_str = f"1-{baggage}"
+    #     pass
+    #     adult_total_price = row['adult_total_price']
+    #     hours_until_departure = row['hours_until_departure']
+        
+    pass
+
+if __name__ == "__main__":
+    follow_up_handle()