263 lines
8.1 KiB
Python
263 lines
8.1 KiB
Python
# 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 []
|