CoolBotDataSys/crawlers/yichens_spider.py

"""一尘网爬虫 - 适配 pm001.net 连体纪念钞板块"""
import re
import httpx
import json
from typing import Optional, Dict, List, Any
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
import logging
import sys

sys.path.insert(0, '/root/coolbot-data')
from crawlers.base import BaseSpider, PaginationSpider
from database import db

logger = logging.getLogger(__name__)


class YichensSpider(PaginationSpider):
    """一尘网/pm001.net 连体纪念钞板块爬虫"""
    
    def __init__(self):
        super().__init__("一尘网连体纪念钞", "pm001")
        self.base_url = "http://www.pm001.net"
        self.board_id = "151"
        self.forum_url = f"{self.base_url}/index.asp?boardid={self.board_id}"
        self.max_pages = 10
        self.encoding = "gbk"
        self.min_delay = 3.0
        self.max_delay = 6.0
    
    def get(self, url: str, **kwargs) -> Optional[Any]:
        import requests
        try:
            self._random_delay()
            headers = {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                "Accept": "text/html,application/xhtml+xml",
                "Accept-Language": "zh-CN,zh;q=0.9",
            }
            response = requests.get(url, timeout=30, headers=headers, **kwargs)
            try:
                response.encoding = "gbk"
            except:
                response.encoding = "gb2312"
            return response
        except Exception as e:
            logger.error(f"请求失败: {e}")
            return None
    
    def parse_posts(self, html: str, url: str) -> List[Dict]:
        """解析帖子列表"""
        posts = []
        pattern = r'<a[^>]*href=["\']?[^"\']*boardID=151[^"\']*ID=(\d+)[^"\']*["\']?[^>]*>([^<]+)</a>'
        matches = re.findall(pattern, html, re.I)
        
        seen_ids = set()
        for post_id, title in matches:
            title = title.strip()
            if len(title) < 5:
                continue
            invalid_titles = ['栏目交易规范', '固', '精华', '_TOP', '锁', '查看', '页']
            if title in invalid_titles:
                continue
            if re.match(r'^\d{4}/\d+/\d+', title):
                continue
            if title.startswith('&nbsp'):
                continue
            if any(k in title for k in ['页', '上一步', '下一步', '发表', '回复']):
                continue
            
            if post_id not in seen_ids:
                seen_ids.add(post_id)
                post_type = 'deal' if any(c in title for c in ['出', '售', '价', '求', '收', '买', '卖', '兑']) else 'normal'
                
                post = {
                    'post_id': post_id,
                    'topic_id': post_id,
                    'title': title,
                    'content': None,
                    'content_html': None,
                    'author_id': f"user_{post_id}",
                    'author_username': '未知',
                    'category': '连体纪念钞',
                    'sub_category': None,
                    'post_type': post_type,
                    'price': self._extract_price(title),
                    'price_unit': self._extract_price_unit(title),
                    'view_count': 0,
                    'reply_count': 0,
                    'like_count': 0,
                    'is_top': False,
                    'is_essence': False,
                    'is_closed': False,
                    'created_at': None,
                    'updated_at': None,
                }
                posts.append(post)
        
        return posts
    
    def parse_post_detail(self, html: str, url: str) -> Optional[Dict]:
        """解析帖子详情页"""
        soup = BeautifulSoup(html, "html.parser")
        
        # 提取帖子ID
        match = re.search(r"ID=(\d+)", url)
        post_id = match.group(1) if match else None
        
        # 从页面文本提取所有关键信息
        page_text = soup.get_text()
        
        # 提取标题 - 从<title>获取
        title = None
        title_elem = soup.find("title")
        if title_elem:
            title_text = title_elem.get_text(strip=True)
            # 格式: "标题[投资资讯网交易在线]"
            if "[" in title_text:
                title = title_text.split("[")[0].strip()
        
        # 提取作者
        author = None
        author_match = re.search(r">>>\s*([^\s<]+)\s*</a>.*?交易等级", page_text, re.DOTALL)
        if author_match:
            author = author_match.group(1).strip()
        else:
            # 尝试另一种模式
            author_patterns = [
                r"([\u4e00-\u9fa5]{2,10}钱币[\u4e00-\u9fa5]*)",  # xxx钱币收藏
                r"([\u4e00-\u9fa5]{2,6}收藏)",  # xxx收藏
                r">>>\s*([^\s<]+)",  # >>>
            ]
            for pattern in author_patterns:
                m = re.search(pattern, page_text)
                if m:
                    author = m.group(1).strip()
                    break
        
        # 提取价格
        price = None
        price_unit = None
        
        # 优先从标题/内容匹配价格
        price_patterns = [
            r'(\d+)\s*元\s*(?:一张|一张通货|一刀|一条|一张单张)',
            r'([\d]+)\s*元\s*出?售',
            r'出?售[^\d]*(\d+)\s*元?',
            r'收[^\d]*(\d+)\s*元?',
            r'(\d+)\s*-\s*(\d+)\s*元',  # 范围价
        ]
        for pattern in price_patterns:
            m = re.search(pattern, page_text)
            if m:
                try:
                    price = float(m.group(1))
                    break
                except:
                    pass
        
        # 提取价格单位
        if '刀' in page_text:
            price_unit = '元/刀'
        elif '张' in page_text:
            price_unit = '元/张'
        elif '条' in page_text:
            price_unit = '元/条'
        elif '套' in page_text:
            price_unit = '元/套'
        else:
            price_unit = '元'
        
        # 提取手机号
        phones = re.findall(r'1[3-9]\d{9,10}', page_text)
        contact = ','.join(dict.fromkeys(phones[:3]))  # 去重，最多3个
        
        # 提取时间 - "2026/4/4 18:15:00"
        created_at = None
        time_match = re.search(r'(\d{4})/(\d{1,2})/(\d{1,2})\s+(\d{1,2}):(\d{2}):(\d{2})', page_text)
        if time_match:
            try:
                dt = datetime(int(time_match.group(1)), int(time_match.group(2)), 
                           int(time_match.group(3)), int(time_match.group(4)),
                           int(time_match.group(5)), int(time_match.group(6)))
                created_at = dt.strftime("%Y-%m-%d %H:%M:%S")
            except:
                pass
        
        # 提取内容（主要文本）
        content = None
        # 找"出售/收购"等关键词后的内容
        content_patterns = [
            r'(出售|收购|求购)[^\n]{10,500}',
            r'(出|售|收)[^\n]{10,500}',
        ]
        for pattern in content_patterns:
            m = re.search(pattern, page_text)
            if m:
                content = m.group(0)[:500]  # 限制长度
                break
        
        return {
            'post_id': post_id,
            'title': title,
            'content': content,
            'author_username': author,
            'price': price,
            'price_unit': price_unit,
            'contact': contact if contact else None,
            'created_at': created_at,
        }
    

    def _extract_price_with_llm(self, title: str) -> dict:
        """Use LLM to extract price from title"""
        prompt = f"""从以下钱币收藏帖子标题中提取信息：
标题："{title}"
提取：post_type(want求购/deal出售/空), price(价格数字或null), unit(元/张、元/刀、元/条、元/套、元等)
回答JSON格式：{{"post_type":"want/deal/","price":数字,"unit":"元/张等"}}"""

        try:
            r = httpx.post(
                "https://coding.dashscope.aliyuncs.com/v1/chat/completions",
                headers={
                    "Authorization": "Bearer sk-sp-d5ce68bb203e48ca857c2aea25255b26",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "qwen3.5-plus",
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.1
                },
                timeout=30.0
            )
            if r.status_code == 200:
                result = r.json()
                content_text = result['choices'][0]['message']['content'].strip()
                if '{' in content_text:
                    json_str = content_text[content_text.find('{'):content_text.rfind('}')+1]
                    return json.loads(json_str)
        except Exception as e:
            print(f"LLM error: {e}")
        return {}

    def _extract_price(self, text: str) -> Optional[float]:
        """从标题提取价格"""
        # 第一优先级：明确的价格单位（元、万）
        patterns_with_unit = [
            r'(\d+(?:\.\d+)?)\s*[万Ww]?\s*元',  # 123元, 1.5万元
            r'[每单共总]\s*(\d+(?:\.\d+)?)\s*元',  # 共123元
            r'(\d+(?:\.\d+)?)\s*元\s*(?:每|每张|每刀|每条|每套)',  # 123元每张
        ]
        for pattern in patterns_with_unit:
            m = re.search(pattern, text)
            if m:
                try:
                    val = float(m.group(1))
                    if val > 0:
                        return val
                except:
                    pass
        
        # 第二优先级：明确标价的（价格X元）
        price_keyword = r'(?:价格|价|售价|收价|成交价)[：:]\s*(\d+(?:\.\d+)?)'
        m = re.search(price_keyword, text)
        if m:
            try:
                val = float(m.group(1))
                if val > 0:
                    return val
            except:
                pass
        
        # 第三优先级：数字+万（万前面肯定是价格）
        wan_pattern = r'(\d+(?:\.\d+)?)\s*万'
        m = re.search(wan_pattern, text)
        if m:
            try:
                val = float(m.group(1))
                if val > 0:
                    return val
            except:
                pass
        
        return None
    
    def _extract_price_unit(self, text: str) -> Optional[str]:
        """从标题提取价格单位"""
        if '刀' in text:
            return '元/刀'
        elif '张' in text:
            return '元/张'
        elif '条' in text:
            return '元/条'
        elif '套' in text:
            return '元/套'
        elif '万' in text:
            return '元/万'
        return '元'
    

    def batch_llm_price_update(self) -> int:
        """批量处理今日缺少价格的帖子，返回更新数量"""
        from database import get_db
        import httpx, json
        
        # 找出今日爬取的价格为空的帖子
        with get_db() as conn:
            cur = conn.cursor()
            cur.execute("""
                SELECT id, title FROM yichens_posts 
                WHERE post_time::date = CURRENT_DATE 
                AND (price IS NULL OR price = 0)
                AND title IS NOT NULL
                LIMIT 50
            """)
            posts = cur.fetchall()
            cur.close()
        
        if not posts:
            print(f"No posts needing LLM price extraction")
            return 0
        
        print(f"Found {len(posts)} posts needing LLM price extraction")
        
        # 批量提取（每批10条）
        batch_size = 10
        updated = 0
        
        for i in range(0, len(posts), batch_size):
            batch = posts[i:i+batch_size]
            
            # 构建批量prompt
            titles_text = '\n'.join([f'{p[0]}|{p[1]}' for p in batch])
            prompt = f"""分析以下钱币收藏帖子的标题，提取实际交易价格。

格式要求：
- 如果有明确价格（数字+元/张、元/刀、元/条、元/套等），提取价格数字
- "求2组"不是价格，是数量
- "765出一组"中765如果是价格则提取
- 如果无法判断实际交易价格，填null

标题列表：
{titles_text}

回答JSON数组格式（只回答JSON，不要其他内容）：
[{{"id":帖子ID,"price":数字或null,"unit":"元/张等"}},...]"""

            try:
                r = httpx.post(
                    "https://coding.dashscope.aliyuncs.com/v1/chat/completions",
                    headers={
                        "Authorization": "Bearer sk-sp-d5ce68bb203e48ca857c2aea25255b26",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": "qwen3.5-plus",
                        "messages": [{"role": "user", "content": prompt}],
                        "temperature": 0.1
                    },
                    timeout=120.0
                )
                
                if r.status_code == 200:
                    result = r.json()
                    content_text = result['choices'][0]['message']['content'].strip()
                    if '[' in content_text:
                        json_str = content_text[content_text.find('['):content_text.rfind(']')+1]
                        results = json.loads(json_str)
                        
                        with get_db() as conn:
                            cur = conn.cursor()
                            for res in results:
                                if res.get('price') and res['price'] > 0:
                                    cur.execute(
                                        "UPDATE yichens_posts SET price = %s, price_unit = %s WHERE id = %s",
                                        (float(res['price']), res.get('unit', '元'), res['id'])
                                    )
                                    updated += 1
                            cur.close()
                        print(f"Batch {i//batch_size + 1}: updated {len(results)} posts")
            except Exception as e:
                print(f"Batch {i//batch_size + 1} error: {e}")
        
        print(f"Total LLM price updates: {updated}")
        return updated


    def save_post(self, post: Dict) -> bool:
        """保存帖子到数据库"""
        if not post or not post.get('post_id'):
            return False
        try:
            with db.get_cursor() as cursor:
                cursor.execute("""
                    INSERT INTO yichens_posts (
                        post_id, topic_id, title, content, content_html,
                        author_id, author_username, category, sub_category,
                        post_type, price, price_unit, view_count, reply_count,
                        like_count, is_top, is_essence, is_closed, created_at, updated_at
                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    ON DUPLICATE KEY UPDATE
                        title = COALESCE(VALUES(title), title),
                        content = COALESCE(VALUES(content), content),
                        author_username = COALESCE(VALUES(author_username), author_username),
                        price = COALESCE(VALUES(price), price),
                        price_unit = COALESCE(VALUES(price_unit), price_unit),
                        created_at = COALESCE(VALUES(created_at), created_at),
                        updated_at = VALUES(updated_at),
                        crawled_at = NOW()
                """, (
                    post.get('post_id'),
                    post.get('topic_id'),
                    post.get('title'),
                    post.get('content'),
                    post.get('content_html'),
                    post.get('author_id'),
                    post.get('author_username'),
                    post.get('category'),
                    post.get('sub_category'),
                    post.get('post_type', 'normal'),
                    post.get('price'),
                    post.get('price_unit'),
                    post.get('view_count', 0),
                    post.get('reply_count', 0),
                    post.get('like_count', 0),
                    post.get('is_top', False),
                    post.get('is_essence', False),
                    post.get('is_closed', False),
                    post.get('created_at'),
                    post.get('updated_at')
                ))
            return True
        except Exception as e:
            logger.error(f"保存帖子失败: {e}")
            return False
    
    def crawl_detail(self, post_id: str) -> Optional[Dict]:
        """爬取单个帖子详情"""
        url = f"http://www.pm001.net/dispbbs.asp?boardID={self.board_id}&ID={post_id}&page=1"
        response = self.get(url)
        if not response:
            return None
        return self.parse_post_detail(response.text, url)
    
    def crawl_forum(self, max_pages: int = 5, crawl_detail: bool = False) -> List[Dict]:
        """爬取板块"""
        logger.info(f"开始爬取连体纪念钞板块: {self.forum_url}")
        self.max_pages = max_pages
        
        all_posts = []
        
        for page in range(1, self.max_pages + 1):
            page_url = f"{self.forum_url}&page={page}"
            logger.info(f"爬取第 {page} 页: {page_url}")
            
            response = self.get(page_url)
            if not response:
                logger.warning(f"第 {page} 页请求失败")
                continue
            
            posts = self.parse_posts(response.text, response.url)
            if not posts:
                logger.info(f"第 {page} 页无数据，停止")
                break
            
            for post in posts:
                if crawl_detail:
                    detail = self.crawl_detail(post['post_id'])
                    if detail:
                        post.update({
                            'title': detail.get('title') or post.get('title'),
                            'content': detail.get('content'),
                            'author_username': detail.get('author_username', '未知'),
                            'price': detail.get('price') or post.get('price'),
                            'price_unit': detail.get('price_unit') or post.get('price_unit'),
                            'created_at': detail.get('created_at'),
                        })
                
                self.save_post(post)
                all_posts.append(post)
            
            logger.info(f"第 {page} 页获取 {len(posts)} 条帖子")
        
        logger.info(f"连体纪念钞板块共采集 {len(all_posts)} 条帖子")
        return all_posts
    
    def run(self, max_pages: int = 5, crawl_detail: bool = False) -> List[Dict]:
        """执行爬虫"""
        log_id = self._log_start()
        
        try:
            posts = self.crawl_forum(max_pages, crawl_detail)
            print(f"Crawled {len(posts)} posts, running LLM price extraction...")
            self.batch_llm_price_update()
            self._log_finish(log_id, "success", len(posts))
            return posts
        except Exception as e:
            logger.error(f"爬虫执行失败: {e}")
            self._log_finish(log_id, "failed", 0, str(e))
            return []
    
    def _log_start(self) -> int:
        with db.get_cursor() as cursor:
            cursor.execute(
                "INSERT INTO crawl_logs (source, status, started_at) VALUES (%s, %s, NOW())",
                (self.source, "running")
            )
            return cursor.lastrowid
    
    def _log_finish(self, log_id: int, status: str, items_count: int, error: str = ""):
        with db.get_cursor() as cursor:
            cursor.execute(
                "UPDATE crawl_logs SET status = %s, items_count = %s, error_message = %s, finished_at = NOW() WHERE id = %s",
                (status, items_count, error, log_id)
            )


def crawl_yichens_lianti(max_pages: int = 5, crawl_detail: bool = False) -> List[Dict]:
    """采集连体纪念钞板块"""
    spider = YichensSpider()
    return spider.run(max_pages, crawl_detail)


if __name__ == "__main__":
    import sys
    max_pages = int(sys.argv[1]) if len(sys.argv) > 1 else 5
    crawl_detail = '--detail' in sys.argv
    crawl_yichens_lianti(max_pages, crawl_detail)