"""一尘网爬虫 - 适配 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']*href=["\']?[^"\']*boardID=151[^"\']*ID=(\d+)[^"\']*["\']?[^>]*>([^<]+)' 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(' '): 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 = 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)