"""只采集今天的新帖子 - 优化版""" import re import sys import os sys.path.insert(0, '/root/coolbot-data') import requests from bs4 import BeautifulSoup from datetime import datetime from crawlers.base import PaginationSpider from database import get_db import logging logger = logging.getLogger(__name__) class YichensTodaySpider(PaginationSpider): 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 = 2 self.encoding = "gbk" self.min_delay = 1.5 self.max_delay = 3.0 def get(self, url, **kwargs): 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,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } response = requests.get(url, timeout=30, headers=headers, **kwargs) try: response.encoding = self.encoding except: response.encoding = "gb2312" return response except Exception as e: print(f"请求失败: {e}") return None def extract_posts_with_dates(self, html): """从列表页提取所有帖子ID和"发表于"时间，返回 [(post_id, date_str), ...] 使用BeautifulSoup解析listtitle div，比正则更准确 """ soup = BeautifulSoup(html, 'html.parser') posts = [] listtitle_divs = soup.find_all('div', class_='listtitle') for div in listtitle_divs: link = div.find('a', href=re.compile(r'dispbbs\.asp\?boardID=151&ID=\d+')) if not link: continue href = link.get('href', '') id_match = re.search(r'\&ID=(\d+)', href) if not id_match: continue post_id = id_match.group(1) title_attr = link.get('title', '') # title格式: 《标题》\n作者：xxx\n发表于：2026/4/5 9:35:00 date_match = re.search(r'发表于：(\d{4}/\d{1,2}/\d{1,2})', title_attr) if date_match: date_str = date_match.group(1) posts.append((post_id, date_str)) return posts def parse_post_detail(self, html, url): """解析详情页，提取真实发帖时间""" soup = BeautifulSoup(html, "html.parser") page_text = soup.get_text() match = re.search(r"ID=(\d+)", url) post_id = match.group(1) if match else None 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 m = re.search(r">>>\s*([^\s<]+)", page_text) if m: author = m.group(1).strip() created_at = None m = re.search(r"(\d{4}/\d{1,2}/\d{1,2}\s+\d{1,2}:\d{2}:\d{2})", page_text) if m: created_at = m.group(1).replace("/", "-") phones = re.findall(r'1[3-9]\d{9,10}', page_text) contact = ",".join(dict.fromkeys(phones[:5])) if phones else None price = None price_unit = None if title: price, price_unit = self.extract_price(title) if not price: price, price_unit = self.extract_price(page_text) content = None m = re.search(r"(出售|收购|求购|出|售|收)[^\n]{10,500}", page_text) if m: content = m.group(0)[:1000] special_types = [] if title: if "救生圈" in title or "88" in title or "888" in title: special_types.append("救生圈") if "标十" in title: special_types.append("标十") if "标百" in title: special_types.append("标百") post_type = "normal" if title: if any(c in title for c in ["出", "售", "卖", "兑"]): post_type = "deal" elif any(c in title for c in ["求", "收", "购"]): post_type = "want" category = "其他" if title: if "龙钞" in title or "龙纪念" in title: category = "龙钞" elif "蛇钞" in title: category = "蛇钞" elif "马钞" in title: category = "马钞" return { "post_id": post_id, "title": title, "content": content, "author_username": author, "price": price, "price_unit": price_unit, "contact": contact, "post_time": created_at, "special_types": "|".join(special_types) if special_types else None, "category": category, "post_type": post_type, "url": f"http://www.pm001.net/dispbbs.asp?boardID={self.board_id}&ID={post_id}&page=1", } def extract_price(self, text): if not text: return None, None patterns = [ r"(\d+)\s*元\s*(?:一张|一刀|一条|单张)?", r"[收售出求兑买]\s*(\d+)\s*元?", ] for pattern in patterns: m = re.search(pattern, text) if m: try: val = float(m.group(1)) if val < 0 or val > 999999999: continue price_unit = "元" if "张" in text: price_unit = "元/张" elif "刀" in text: price_unit = "元/刀" elif "条" in text: price_unit = "元/条" return val, price_unit except: pass return None, None def crawl_today(self): """全量采集：爬取第1-max_pages页所有帖子，不过滤日期 用于一次性补全所有帖子的post_time等字段 """ today = datetime.now().date() today_str = today.strftime("%Y/%m/%d").lstrip('0').replace('/0', '/') print(f"开始全量采集第1-{self.max_pages}页 (今天: {today_str})") saved = 0 checked = 0 for page in range(1, self.max_pages + 1): page_url = f"{self.forum_url}&page={page}" print(f"\n扫描第 {page} 页...") response = self.get(page_url) if not response: continue posts = self.extract_posts_with_dates(response.text) print(f" 该页共 {len(posts)} 条帖子") for post_id, date_str in posts: checked += 1 # 标准化日期 normalized_date = date_str.lstrip('0').replace('/0', '/') # 访问详情页 detail_url = f"{self.base_url}/dispbbs.asp?boardID={self.board_id}&ID={post_id}&page=1" detail_response = self.get(detail_url) if not detail_response: continue detail = self.parse_post_detail(detail_response.text, detail_url) # 验证详情页的真实发帖时间 post_time_str = detail.get('post_time', '') if not post_time_str: continue try: post_time = datetime.strptime(post_time_str, '%Y-%m-%d %H:%M:%S').date() is_today = (post_time == today) label = f"今日{normalized_date}" if is_today else f"历史{post_time}" except: label = "时间异常" # 保存（ON CONFLICT会更新post_time等字段） if self.save_full_post(detail): saved += 1 print(f" [保存] {post_id} - {detail.get('title', '')[:25]}... ({label}中已存{saved}条)") else: print(f" [失败] {post_id}") print(f" 页码 {page} 完成: 检查{len(posts)}条") print(f"\n===== 采集完成 =====") print(f"总计检查: {checked} 条") print(f"成功保存: {saved} 条") return saved def save_full_post(self, post): if not post or not post.get("post_id"): return False sql = """ INSERT INTO yichens_posts ( post_id, title, content, category, post_type, price, price_unit, special_types, number_features, author_username, author_id, contact, has_lifebuoy, reply_count, view_count, post_time, crawled_at, updated_at, url ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s) ON CONFLICT (post_id) DO UPDATE SET title = EXCLUDED.title, content = EXCLUDED.content, category = EXCLUDED.category, post_type = EXCLUDED.post_type, price = EXCLUDED.price, price_unit = EXCLUDED.price_unit, special_types = EXCLUDED.special_types, author_username = EXCLUDED.author_username, contact = EXCLUDED.contact, has_lifebuoy = EXCLUDED.has_lifebuoy, post_time = EXCLUDED.post_time, updated_at = NOW(), crawled_at = NOW(), url = EXCLUDED.url """ has_lifebuoy = post.get("special_types") and "救生圈" in post.get("special_types", "") try: with get_db() as conn: if not conn: print("数据库连接失败") return False cur = conn.cursor() cur.execute(sql, ( post.get("post_id"), post.get("title"), post.get("content"), post.get("category"), post.get("post_type"), post.get("price"), post.get("price_unit"), post.get("special_types"), None, post.get("author_username"), f"user_{post.get('post_id')}", post.get("contact"), has_lifebuoy, 0, 0, post.get("post_time"), post.get("url"), )) conn.commit() cur.close() return True except Exception as e: print(f"保存失败: {e}") return False def run(self): log_id = self._log_start() try: result = self.crawl_today() self._log_finish(log_id, "success", result) return result except Exception as e: print(f"爬虫异常: {e}") self._log_finish(log_id, "failed", 0, str(e)) return 0 def _log_start(self): try: with get_db() as conn: cur = conn.cursor() cur.execute( "INSERT INTO crawl_logs (spider_name, status, started_at) VALUES (%s, %s, NOW()) RETURNING id", (self.source, "running") ) log_id = cur.fetchone()[0] conn.commit() cur.close() return log_id except Exception as e: print(f"记录日志失败: {e}") return None def _log_finish(self, log_id, status, items_count, error=""): if log_id is None: return try: with get_db() as conn: cur = conn.cursor() cur.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) ) conn.commit() cur.close() except Exception as e: print(f"更新日志失败: {e}") if __name__ == "__main__": spider = YichensTodaySpider() spider.run()