# coolbot_matcher - 一尘数据库号码匹配工具 # Version: 0.0.2 (2026-04-30) # 描述:从一尘数据库(coolbot_data)查询匹配的藏品号码 # 更新:号码分类规则修正 from typing import List, Optional from sqlalchemy import text from sqlalchemy.orm import Session from app.core.coolbot_db import coolbot_engine def classify_number(number: str) -> str: """根据号码特征分类 规则(2026-04-30修正版): 首先区分标百、标十,单张: - 单张:J后面9位 - 标十:J后面8位(最后一位是1) - 标百:J后面7位(最后两位是01) 分类优先级: | 类型 | 排除 | 可用数字 | 必须包含 | |------|------|----------|----------| | 通货 | - | - | 4 | | 带7号 | 4 | - | 7 | | 永恒号 | 47 | - | - | | 圆圆号 | 123457 | 0689 | - | | 倒置号 | 23457 | 01689 | 1 | | 金马王 | 12347 | 05689 | 5 | | 金马号 | 2347 | 015689 | 1,5 | | 金山王 | 12457 | 03689 | 3 | | 天马王 | 1247 | 035689 | 3,5 | | 金山号 | 2457 | 013689 | 1,3 | | 天马号 | 247 | 0135689 | 1,3,5 | | 朦胧王 | 13457 | - | - | | 朦胧号 | 3457 | - | - | | 如意号 | 1347 | - | - | | 钻石号 | 347 | - | - | """ if not number: return "未知" if number.startswith('J0'): digits = number[2:] elif number.startswith('J'): digits = number[1:] else: return "未知" if len(digits) == 9: d = digits elif len(digits) == 8: d = digits[:8] elif len(digits) == 7: d = digits else: return "未知" unique = set(d) # 1. 通货:带4 if '4' in unique: return "通货" # 2. 带7号:无4,有7 if '7' in unique: return "带7号" # 3. 圆圆号:无123457,0689四个数字任意组合 if unique <= {'0', '6', '8', '9'}: return "圆圆号" # 4. 永恒号:只有0和1(无47但不符合圆圆号/倒置号) if unique <= {'0', '1'}: return "永恒号" # 5. 倒置号:01689组合,必须有1,且包含6/8/9中的至少一个 if unique <= {'0', '1', '6', '8', '9'} and '1' in unique and unique & {'6', '8', '9'}: return "倒置号" # 6. 金马王:无12347,05689组合,必须有5(排除1) if unique <= {'0', '5', '6', '8', '9'} and '5' in unique and '1' not in unique: return "金马王" # 7. 金马号:无2347,015689组合,必须有1和5(包含1) if unique <= {'0', '1', '5', '6', '8', '9'} and '1' in unique and '5' in unique: return "金马号" # 8. 金山王:无12457,03689组合,必须有3(排除1和5) if unique <= {'0', '3', '6', '8', '9'} and '3' in unique and '1' not in unique and '5' not in unique: return "金山王" # 9. 天马王:无1247,035689组合,必须有3和5(排除1) if unique <= {'0', '3', '5', '6', '8', '9'} and '3' in unique and '5' in unique and '1' not in unique: return "天马王" # 10. 金山号:无2457,013689组合,必须有1和3(排除5) if unique <= {'0', '1', '3', '6', '8', '9'} and '1' in unique and '3' in unique and '5' not in unique: return "金山号" # 11. 天马号:无247,0135689组合,必须有1、3和5 if unique <= {'0', '1', '3', '5', '6', '8', '9'} and '1' in unique and '3' in unique and '5' in unique: return "天马号" # 12. 朦胧王:无13457 if not unique & {'1', '3', '4', '5', '7'}: return "朦胧王" # 13. 朦胧号:无3457 if not unique & {'3', '4', '5', '7'}: return "朦胧号" # 14. 如意号:无1347 if not unique & {'1', '3', '4', '7'}: return "如意号" # 15. 钻石号:无347 if not unique & {'3', '4', '7'}: return "钻石号" # 16. 永恒号兜底:无47 if '7' not in unique: return "永恒号" return "其他" def check_match(col_number: str, expect_number: str, expect_category: str) -> bool: """检查藏品号码是否符合期望的分类""" if not col_number or not expect_category: return False col_cat = classify_number(col_number) if col_cat == expect_category: return True if expect_category == "其他": return check_number_pattern(col_number, expect_number) return False def check_number_pattern(col_number: str, expect_number: str) -> bool: """检查号码特征匹配""" if not col_number or not expect_number: return False if col_number.startswith('J0'): col_digits = col_number[2:] else: col_digits = col_number[1:] if len(col_number) > 1 else col_number if expect_number.startswith('J0'): exp_digits = expect_number[2:] else: exp_digits = expect_number[1:] if len(expect_number) > 1 else expect_number min_len = min(len(col_digits), len(exp_digits)) return col_digits[:min_len] == exp_digits[:min_len] def get_match_type(expect_number: str) -> str: """判断匹配类型""" if not expect_number: return "unknown" if expect_number.startswith('J0'): digits = expect_number[2:] elif expect_number.startswith('J'): digits = expect_number[1:] else: return "unknown" if len(digits) == 8: return "ten" elif len(digits) == 7: return "hundred" elif len(digits) == 9: return "single" return "unknown" def match_self_collections_count(db: Session, user_id: str, expect_number: str, expect_category: str) -> int: """计算匹配藏品数量(用户自有藏品)""" from app.models.models import Collection if not expect_number: return 0 collections = db.query(Collection).filter( Collection.f99_91_user_id == user_id, Collection.f01_04_status == "in_collection" ).all() count = 0 for c in collections: number = c.f02_10_prefix_serial or '' if check_match(number, expect_number, expect_category): count += 1 return count def match_collections_count_from_coolbot(expect_number: str, expect_category: str) -> int: """计算匹配藏品数量(一尘数据库)""" if not expect_number: return 0 query = text(""" SELECT id, crown_code FROM collections WHERE crown_code IS NOT NULL AND crown_code != '' AND crown_code LIKE 'J0%' """) try: with coolbot_engine.connect() as conn: result = conn.execute(query) return sum(1 for row in result if check_match(row[1], expect_number, expect_category)) except Exception as e: print(f"Error: {e}") return 0 def match_collections_list_from_coolbot(expect_number: str, expect_category: str, limit: int = 20) -> List[dict]: """获取匹配的藏品列表(一尘数据库)""" if not expect_number: return [] query = text(""" SELECT id, name, category, crown_code, price, post_title, post_url, author, post_crawled_at FROM collections WHERE crown_code IS NOT NULL AND crown_code != '' AND crown_code LIKE 'J0%' """) try: with coolbot_engine.connect() as conn: result = conn.execute(query) matched = [] for row in result: if check_match(row[3], expect_number, expect_category): matched.append({ "id": row[0], "name": row[1], "category": row[2], "crown_code": row[3], "price": float(row[4]) if row[4] else None, "post_title": row[5], "post_url": row[6], "author": row[7], "post_crawled_at": row[8].isoformat() if row[8] else None }) if len(matched) >= limit: break return matched except Exception as e: print(f"Error: {e}") return []