CoolBotDataSys/scripts/crown_extract_incremental.py

#!/usr/bin/env python3
"""
冠字号全量提取脚本 - 跑完所有待处理帖子
"""
import sys
sys.path.insert(0, '/root/coolbot-data')

import re, psycopg2
from datetime import datetime

DB_CONFIG = {
    'host': 'pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com',
    'port': 5432,
    'database': 'coolbot_data',
    'user': 'coolbot',
    'password': 'Coolbot123'
}

FEATURE_KEYWORDS = [
    '无47', '无34', '无347', '无4', '无3', '无2',
    '标十', '标百', '标九', '标准十', '标准百',
    '首日', '生日', '金马', '银马',
    '金钩', '倒置', '满号', '圆圆',
    '豹子', '顺子', '恐龙', '天龙',
    'PMG', '爱藏', '尾号',
    '大象', '麒麟', '老虎', '狮子',
    '金马王', '天马', '龙马精神',
    '连号', '散号', '一刀', '龙凤', '熊猫',
]

def is_date_code(code):
    if len(code) != 8:
        return False
    try:
        year = int(code[:4])
        month = int(code[4:6])
        day = int(code[6:8])
        if 2000 <= year <= 2030 and 1 <= month <= 12 and 1 <= day <= 31:
            return True
    except:
        pass
    return False

def is_phone(code):
    return len(code) == 11 and code.startswith('1')

def is_valid_crown(code):
    if not code:
        return False
    if code.startswith('J0'):
        return len(code) in (10, 11)
    if len(code) not in (8, 9):
        return False
    if is_phone(code):
        return False
    if len(code) == 8 and is_date_code(code):
        return False
    return True

def extract_codes(text):
    if not text:
        return set()
    codes = set()
    for p in [r'J0\d{8}', r'J0\d{9}']:
        codes.update(re.findall(p, text))
    for c in re.findall(r'\b\d{8}\b', text):
        if is_valid_crown(c):
            codes.add(c)
    for c in re.findall(r'\b\d{9}\b', text):
        if is_valid_crown(c):
            codes.add(c)
    return codes

def extract_features(text):
    if not text:
        return None
    feat = [kw for kw in FEATURE_KEYWORDS if kw in text]
    return '|'.join(feat) if feat else None

def extract_price(text):
    if not text:
        return None, '元'
    prices = []
    for p in [r'(\d+(?:\.\d+)?)\s*元', r'(\d+(?:\.\d+)?)\s*/\s*[张件组百]']:
        for x in re.findall(p, text):
            try:
                v = float(x)
                if 1 <= v < 100000:
                    prices.append(v)
            except:
                pass
    return (min(prices), '元') if prices else (None, '元')

def get_category(title):
    if not title:
        return '其他'
    for kw in ['龙', '龙钞', '小龙钞', '钞王']:
        if kw in title:
            return '龙钞'
    for kw in ['马', '马钞']:
        if kw in title:
            return '马钞'
    for kw in ['蛇', '蛇钞']:
        if kw in title:
            return '蛇钞'
    return '其他'

def main():
    print(f'[{datetime.now()}] 冠字号全量提取开始...')

    conn = psycopg2.connect(**DB_CONFIG)
    cur = conn.cursor()

    # 无 LIMIT，全量跑完所有待处理帖子
    cur.execute("""
        SELECT p.id, p.post_id, p.title, p.content, p.post_type,
               p.author_username, p.price, p.price_unit, p.url, p.crawled_at
        FROM yichens_posts p
        WHERE p.category IN ('龙钞', '马钞', '蛇钞', '其他')
          AND NOT EXISTS (
              SELECT 1 FROM collections c
              WHERE c.post_id = CAST(p.id AS TEXT)
          )
        ORDER BY p.id
    """)
    posts = cur.fetchall()
    total = len(posts)
    print(f'待处理新帖子: {total} 条')

    new_count = 0
    skip_count = 0
    total_codes = 0
    processed = 0

    for (pid, post_id, title, content, post_type,
         author, price, price_unit, url, crawled_at) in posts:

        text = f'{title or ""} {content or ""}'
        codes = extract_codes(text)
        total_codes += len(codes)
        processed += 1

        if not codes:
            skip_count += 1
            if processed % 500 == 0:
                print(f'  已处理 {processed}/{total} 条，当前新增 {new_count} 条...')
            continue

        features = extract_features(text)
        price_val, _ = extract_price(text)
        price_val = price_val if price_val else price
        category = get_category(title)

        for code in codes:
            try:
                cur.execute("""
                    INSERT INTO collections (
                        name, crown_code, category, post_id, post_url,
                        author, price, price_unit,
                        number_feature, post_title, post_type,
                        post_crawled_at, created_at, updated_at
                    ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW(),NOW())
                    ON CONFLICT (post_id, crown_code) DO NOTHING
                """, (
                    code, code, category, str(post_id), url,
                    author, price_val, price_unit or '元',
                    features, title, post_type,
                    crawled_at
                ))
                if cur.rowcount > 0:
                    new_count += 1
            except Exception as e:
                print(f'  插入失败 post_id={post_id}, code={code}: {e}')

        if processed % 500 == 0:
            print(f'  已处理 {processed}/{total} 条，当前新增 {new_count} 条...')
            conn.commit()

    conn.commit()
    print(f'完成！新增: {new_count} 条, 总冠号: {total_codes}, 无冠号跳过: {skip_count} 条, 总处理: {total} 条')

    cur.execute('SELECT COUNT(*) FROM collections')
    print(f'collections表当前总量: {cur.fetchone()[0]} 条')
    conn.close()

if __name__ == '__main__':
    main()