jiachenlong/backend/app/utils/coolbot_matcher.py

# coolbot_matcher - 一尘数据库号码匹配工具
# Version: 0.0.1 (2026-04-24)
# 描述：从一尘数据库(coolbot_data)查询匹配的藏品号码

from typing import List
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.core.coolbot_db import coolbot_engine


def match_pattern(col_number: str, pattern: str) -> bool:
    """匹配号码特征模式
    
    通配符规则:
    - X = 任意数字
    - A = 非4
    - B = 非47
    - C = 非347
    - D = 非247
    - E = 非2347
    """
    if not col_number or not pattern:
        return False
    
    if len(col_number) != len(pattern):
        return False
    
    for i, p in enumerate(pattern):
        c = col_number[i]
        if p == 'X':
            continue  # 任意数字
        elif p == 'A':
            if c == '4':
                return False
        elif p == 'B':
            if c == '4':
                return False
            if i < len(col_number) - 1 and col_number[i+1] == '7' and pattern[i+1] == 'X':
                return False
        elif p == 'C':
            if c in '347':
                return False
        elif p == 'D':
            if c in '247':
                return False
        elif p == 'E':
            if c in '2347':
                return False
        elif p == 'L':
            # L = 带4
            if c != '4':
                return False
        elif p == 'N':
            # N = 无4
            if c == '4':
                return False
        else:
            if c != p:
                return False
    return True


def match_self_collections_count(db: Session, user_id: str, expect_number: str) -> int:
    """根据号码特征计算匹配藏品数量（从用户自有藏品）
    
    Args:
        db: 数据库会话
        user_id: 用户ID
        expect_number: 期望号码，如 "J012345678" (固定J0前缀+8位特征)
    
    Returns:
        匹配的藏品数量
    """
    from app.models.models import Collection
    
    if not expect_number or len(expect_number) != 10:
        return 0
    
    if not expect_number.startswith('J0'):
        return 0
    
    pattern = expect_number[2:]  # 后8位
    if not pattern:
        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 len(number) >= 10 and number.startswith('J0'):
            col_pattern = number[2:10]
            if match_pattern(col_pattern, pattern):
                count += 1
        elif len(number) >= 8:
            col_pattern = number[:8]
            if match_pattern(col_pattern, pattern):
                count += 1
    
    return count


def match_collections_count_from_coolbot(expect_number: str) -> int:
    """根据号码特征计算匹配藏品数量（从coolbot_data数据库）
    
    Args:
        expect_number: 期望号码，如 "J012345678" (固定J0前缀+8位特征)
    
    Returns:
        匹配的藏品数量
    """
    if not expect_number or len(expect_number) != 10:
        return 0
    
    if not expect_number.startswith('J0'):
        return 0
    
    pattern = expect_number[2:]  # 后8位
    if not pattern:
        return 0
    
    query = text("""
        SELECT id, crown_code FROM collections 
        WHERE crown_code IS NOT NULL 
        AND crown_code != ''
        AND LENGTH(crown_code) >= 10
        AND crown_code LIKE 'J0%'
    """)
    
    try:
        with coolbot_engine.connect() as conn:
            result = conn.execute(query)
            
            match_count = 0
            for row in result:
                crown_code = row[1]
                if crown_code and len(crown_code) >= 10:
                    col_pattern = crown_code[2:10]
                    if match_pattern(col_pattern, pattern):
                        match_count += 1
            
            return match_count
    except Exception as e:
        print(f"Error querying coolbot_data: {e}")
        return 0


def match_collections_list_from_coolbot(expect_number: str, limit: int = 20) -> List[dict]:
    """获取匹配的藏品列表（从coolbot_data数据库）
    
    Args:
        expect_number: 期望号码，如 "J012345678"
        limit: 返回的最大数量
    
    Returns:
        匹配的藏品列表
    """
    if not expect_number or len(expect_number) != 10:
        return []
    
    if not expect_number.startswith('J0'):
        return []
    
    pattern = expect_number[2:]  # 后8位
    if not pattern:
        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 LENGTH(crown_code) >= 10
        AND crown_code LIKE 'J0%'
    """)
    
    try:
        with coolbot_engine.connect() as conn:
            result = conn.execute(query)
            
            matched = []
            for row in result:
                crown_code = row[3]
                if crown_code and len(crown_code) >= 10:
                    col_pattern = crown_code[2:10]
                    if match_pattern(col_pattern, pattern):
                        matched.append({
                            "id": row[0],
                            "name": row[1],
                            "category": row[2],
                            "crown_code": crown_code,
                            "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 querying coolbot_data: {e}")
        return []