From 93295db357b10c16c52622a3e47ab6b55feea26d Mon Sep 17 00:00:00 2001 From: bizwings Date: Thu, 9 Jul 2026 08:23:21 +0800 Subject: [PATCH] =?UTF-8?q?refactor(BIZ-87):=20P2/P3=20=E8=BF=AD=E4=BB=A3?= =?UTF-8?q?=E2=80=94=E2=80=94=E4=BB=A3=E7=A0=81=E8=B4=A8=E9=87=8F=E6=94=B9?= =?UTF-8?q?=E8=BF=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P2-1: 删除死代码 load_latest_draw() 和 find_next_draw_after() - 两个函数在修复后已无任何调用点，删除避免误导维护者 P2-2: 抽取 row_to_draw 为模块级 _row_to_draw() 公共函数 - 消除三处重复代码，统一调用入口 P2-3: compare_type 细化为 same_day / cross_day / waiting - 原 'current' 语义模糊，现区分同日 vs 跨日归属 P2-4: 边界条件修复 <= → < - 严格对齐规则 2.2：生成时间 ≥ 20:00 归属下一期 - 恰好 20:00:00 现在正确归属下一期 P3: 删除不可达的 elif 死分支 - 移除注释已说明不会到达的防御性代码 验证：8 个测试用例全部通过（含 20:00:00 边界） Co-authored-by: multica-agent --- app.py | 163 ++++++++++++++------------------------------------------- 1 file changed, 38 insertions(+), 125 deletions(-) diff --git a/app.py b/app.py index efff22c..407cc3c 100644 --- a/app.py +++ b/app.py @@ -117,104 +117,47 @@ def load_history_dataframe(): return _load_history(CONFIG['history_file']) -def load_latest_draw(): - """加载最近一期开奖数据，返回 dict 或 None。""" +def _row_to_draw(row): + """将 DataFrame 行转换为开奖数据 dict（公共函数）。 + + 供 find_draw_by_gen_time 及其他需要解析开奖记录的函数统一调用。 + """ + s = str(row.get('号码', '')).strip() + reds, blue = parse_number_string(s) + draw_date_str = str(row.get('开奖时间', '')).strip() + draw_date = None try: - df = load_history_dataframe() - if df is None or df.empty: - return None - - # 解析日期，按开奖时间降序找最近一期 - df['_dt'] = pd.to_datetime(df['开奖时间'], errors='coerce') - df = df.sort_values('_dt', ascending=False).reset_index(drop=True) - - if df.empty: - return None - - row = df.iloc[0] - s = str(row.get('号码', '')).strip() - reds, blue = parse_number_string(s) - - draw_date_str = str(row.get('开奖时间', '')).strip() - draw_date = None + draw_date = datetime.strptime(draw_date_str[:10], '%Y-%m-%d') + except (ValueError, TypeError): try: - draw_date = datetime.strptime(draw_date_str[:10], '%Y-%m-%d') - except (ValueError, TypeError): - try: - dt = pd.to_datetime(draw_date_str, errors='coerce') - if pd.notna(dt): - draw_date = dt.to_pydatetime() - except Exception: - pass - - return { - 'date': draw_date_str, - 'date_obj': draw_date, - 'period': str(row.get('期数', '')).strip(), - 'reds': reds, - 'blue': blue, - 'raw': {k: str(row.get(k, '')).strip() for k in ['开奖时间', '期数', '号码', '和值特征', '奇偶比', '大小比', '跨度']} - } - except Exception as e: - traceback.print_exc() - return None - - -def find_next_draw_after(date_obj): - """查找给定日期之后最近一期的开奖数据，返回 dict 或 None。""" - try: - df = load_history_dataframe() - if df is None or df.empty: - return None - - df['_dt'] = pd.to_datetime(df['开奖时间'], errors='coerce') - # 找开奖日期 > date_obj 的最近一期 - future = df[df['_dt'] > date_obj].sort_values('_dt', ascending=True) - if future.empty: - return None - - row = future.iloc[0] - s = str(row.get('号码', '')).strip() - reds, blue = parse_number_string(s) - - draw_date_str = str(row.get('开奖时间', '')).strip() - draw_date = None - try: - draw_date = datetime.strptime(draw_date_str[:10], '%Y-%m-%d') - except (ValueError, TypeError): - try: - dt = pd.to_datetime(draw_date_str, errors='coerce') - if pd.notna(dt): - draw_date = dt.to_pydatetime() - except Exception: - pass - - return { - 'date': draw_date_str, - 'date_obj': draw_date, - 'period': str(row.get('期数', '')).strip(), - 'reds': reds, - 'blue': blue, - 'raw': {k: str(row.get(k, '')).strip() for k in ['开奖时间', '期数', '号码', '和值特征', '奇偶比', '大小比', '跨度']} - } - except Exception as e: - traceback.print_exc() - return None + dt = pd.to_datetime(draw_date_str, errors='coerce') + if pd.notna(dt): + draw_date = dt.to_pydatetime() + except Exception: + pass + return { + 'date': draw_date_str, + 'date_obj': draw_date, + 'period': str(row.get('期数', '')).strip(), + 'reds': reds, + 'blue': blue, + 'raw': {k: str(row.get(k, '')).strip() for k in ['开奖时间', '期数', '号码', '和值特征', '奇偶比', '大小比', '跨度']} + } def find_draw_by_gen_time(gen_time): """根据生成时间找到归属的开奖期。 比对规则（窗口归属模型）： - - 每个开奖日 D 有一个归属窗口 (D_prev 20:00, D 20:00] + - 每个开奖日 D 有一个归属窗口 (D_prev 20:00, D 20:00) - 生成时间 T 落在哪个窗口内 → 与该开奖日 D 比对 - - 简化实现：按开奖日升序遍历，找到第一个 D 使得 T ≤ D 20:00 - - 如果 T 同日且 T < 20:00 → 同日比对（规则 2.1） - - 如果 T 在 D 之前某日 → 仍归属 D（窗口归属） + - 按开奖日升序遍历，找到第一个 D 使得 T < D 20:00 + - 如果 T 同日且 T < 20:00 → 同日比对（规则 2.1），compare_type='same_day' + - 如果 T 在 D 之前某日 → 仍归属 D（窗口归属），compare_type='cross_day' - 如果 T 同日且 T ≥ 20:00 → 归属下一个开奖日（规则 2.2） - - 如果 T 超过所有开奖日的 20:00 → 等待下一期 + - 如果 T 超过所有开奖日的 20:00 → 等待下一期，compare_type='waiting' - 返回 (draw_dict, compare_type) 或 (None, 'waiting') + 返回 (draw_dict, compare_type) 或 (None, 'error') """ try: df = load_history_dataframe() @@ -227,55 +170,25 @@ def find_draw_by_gen_time(gen_time): if df.empty: return None, 'no_data' - def row_to_draw(row): - s = str(row.get('号码', '')).strip() - reds, blue = parse_number_string(s) - draw_date_str = str(row.get('开奖时间', '')).strip() - draw_date = None - try: - draw_date = datetime.strptime(draw_date_str[:10], '%Y-%m-%d') - except (ValueError, TypeError): - try: - dt = pd.to_datetime(draw_date_str, errors='coerce') - if pd.notna(dt): - draw_date = dt.to_pydatetime() - except Exception: - pass - return { - 'date': draw_date_str, - 'date_obj': draw_date, - 'period': str(row.get('期数', '')).strip(), - 'reds': reds, - 'blue': blue, - 'raw': {k: str(row.get(k, '')).strip() for k in ['开奖时间', '期数', '号码', '和值特征', '奇偶比', '大小比', '跨度']} - } - - # 遍历所有开奖日（升序），找到第一个 D 使得 gen_time ≤ D 20:00 + # 遍历所有开奖日（升序），找到第一个 D 使得 gen_time < D 20:00 for i in range(len(df)): draw_date = df.iloc[i]['_dt'] draw_8pm = draw_date.replace(hour=20, minute=0, second=0, microsecond=0) - if gen_time <= draw_8pm: + if gen_time < draw_8pm: # gen_time 落在 D 的窗口内 → 与 D 比对 - draw = row_to_draw(df.iloc[i]) + draw = _row_to_draw(df.iloc[i]) - # 判断是同日还是窗口归属 - gen_date_only = gen_time.date() - draw_date_only = draw_date.date() - - if gen_date_only == draw_date_only and gen_time < draw_8pm: - compare_type = 'current' # 同日，20:00 前 - elif gen_date_only == draw_date_only and gen_time >= draw_8pm: - # 同日 20:00 后 → 应该归属下一期（但上面条件已判断 ≤，所以这种情况不会进到这里） - # 这个分支在逻辑上不会到达，但保留防御性代码 - continue + # 判断是同日还是跨日窗口归属 + if gen_time.date() == draw_date.date(): + compare_type = 'same_day' # 同日，20:00 前 else: - compare_type = 'current' # 窗口归属（生成时间在 D 之前某个时间点，归属 D） + compare_type = 'cross_day' # 跨日窗口归属 return draw, compare_type # gen_time 超过所有开奖日的 20:00 → 等待下一期 - last_draw = row_to_draw(df.iloc[-1]) + last_draw = _row_to_draw(df.iloc[-1]) return last_draw, 'waiting' except Exception as e: traceback.print_exc()