529 lines
20 KiB
Python
529 lines
20 KiB
Python
"""一尘网爬虫 - 适配 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(' '):
|
||
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)
|