feat: 完成一尘数据采集完整功能
- 一尘爬虫: 列表页解析 + 详情页解析 - 帖子详情: 提取标题/价格/联系方式/时间/内容 - 定时任务: 每天8点/18点自动爬取 - 日报推送: 每天9点推送飞书日报 - 一尘数据: 1800+ 帖子已入库
This commit is contained in:
parent
63a5b46791
commit
c4e029a20a
|
|
@ -135,13 +135,13 @@ async def get_statistics():
|
|||
})
|
||||
|
||||
# 爬虫调度接口
|
||||
from crawlers.yichens_spider import YichensPostSpider
|
||||
from crawlers.yichens_spider import YichensSpider
|
||||
|
||||
@app.post("/api/v1/crawl-jobs/trigger", response_model=ApiResponse, tags=["crawl"])
|
||||
async def trigger_crawl(source: str = "yichens"):
|
||||
"""触发爬虫任务"""
|
||||
if source == "yichens":
|
||||
spider = YichensPostSpider()
|
||||
spider = YichensSpider()
|
||||
items = spider.run()
|
||||
return ApiResponse.success({
|
||||
"source": source,
|
||||
|
|
|
|||
|
|
@ -1,311 +1,347 @@
|
|||
"""一尘网爬虫 - 一尘网钱币论坛数据采集"""
|
||||
"""一尘网爬虫 - 适配 pm001.net 连体纪念钞板块"""
|
||||
import re
|
||||
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 YichensUserSpider(PaginationSpider):
|
||||
"""一尘网用户爬虫"""
|
||||
class YichensSpider(PaginationSpider):
|
||||
"""一尘网/pm001.net 连体纪念钞板块爬虫"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__("一尘网用户", "yichens")
|
||||
self.base_url = "https://www.yichens.com/user"
|
||||
self.user_list_url = "https://www.yichens.com/user/list"
|
||||
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 parse_user(self, html: str, url: str) -> Optional[Dict]:
|
||||
soup = BeautifulSoup(html, "lxml")
|
||||
user_id = None
|
||||
match = re.search(r"user[_\-]?id[=:\s]*['\"]?(\w+)", url, re.I)
|
||||
if match:
|
||||
user_id = match.group(1)
|
||||
|
||||
match = re.search(r"/user/([^/]+)", url)
|
||||
username = match.group(1) if match else None
|
||||
|
||||
if not user_id and not username:
|
||||
return None
|
||||
|
||||
user = {
|
||||
"user_id": user_id or username,
|
||||
"username": username,
|
||||
"nickname": None,
|
||||
"avatar_url": None,
|
||||
"user_level": None,
|
||||
"credit_score": 0,
|
||||
"register_date": None,
|
||||
"last_active_at": None,
|
||||
"is_seller": False,
|
||||
"seller_rating": None,
|
||||
"is_verified": False,
|
||||
"bio": None,
|
||||
"province": None,
|
||||
}
|
||||
return user
|
||||
|
||||
def parse_posts(self, html: str, url: str) -> List[Dict]:
|
||||
return []
|
||||
|
||||
def parse_post_detail(self, html: str, url: str) -> Optional[Dict]:
|
||||
return None
|
||||
|
||||
def crawl_user_detail(self, user_id: str) -> Optional[Dict]:
|
||||
url = f"{self.base_url}/{user_id}"
|
||||
response = self.get(url)
|
||||
if not response:
|
||||
return None
|
||||
return self.parse_user(response.text, url)
|
||||
|
||||
def save_user(self, user: Dict) -> bool:
|
||||
if not user or not user.get("user_id"):
|
||||
return False
|
||||
def get(self, url: str, **kwargs) -> Optional[Any]:
|
||||
import requests
|
||||
try:
|
||||
with db.get_cursor() as cursor:
|
||||
cursor.execute("""
|
||||
INSERT INTO yichens_users (user_id, username, nickname, avatar_url, user_level,
|
||||
credit_score, register_date, last_active_at, is_seller,
|
||||
seller_rating, is_verified, bio, province)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
ON DUPLICATE KEY UPDATE username = VALUES(username),
|
||||
nickname = VALUES(nickname), crawl_latest_at = NOW()
|
||||
""", (
|
||||
user.get("user_id"), user.get("username"), user.get("nickname"),
|
||||
user.get("avatar_url"), user.get("user_level"), user.get("credit_score", 0),
|
||||
user.get("register_date"), user.get("last_active_at"), user.get("is_seller", False),
|
||||
user.get("seller_rating"), user.get("is_verified", False),
|
||||
user.get("bio"), user.get("province")
|
||||
))
|
||||
logger.info(f"用户保存成功: {user.get('username')}")
|
||||
return True
|
||||
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 False
|
||||
|
||||
def run(self) -> List[Dict]:
|
||||
logger.info("开始采集一尘网用户...")
|
||||
return []
|
||||
|
||||
|
||||
class YichensPostSpider(PaginationSpider):
|
||||
"""一尘网帖子爬虫"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__("一尘网帖子", "yichens")
|
||||
self.base_url = "https://www.yichens.com"
|
||||
self.forum_url = "https://www.yichens.com/forum"
|
||||
self.max_pages = 5
|
||||
self.categories = {
|
||||
"longchao": {"name": "龙钞", "url": "/forum/longchao"},
|
||||
"snake": {"name": "蛇钞", "url": "/forum/snake"},
|
||||
"horse": {"name": "马钞", "url": "/forum/horse"},
|
||||
}
|
||||
logger.error(f"请求失败: {e}")
|
||||
return None
|
||||
|
||||
def parse_posts(self, html: str, url: str) -> List[Dict]:
|
||||
soup = BeautifulSoup(html, "lxml")
|
||||
"""解析帖子列表"""
|
||||
posts = []
|
||||
post_items = soup.select(".topic-item, .post-item, .thread-item")
|
||||
for item in post_items:
|
||||
try:
|
||||
post = self._extract_post(item, url)
|
||||
if post:
|
||||
posts.append(post)
|
||||
except Exception as e:
|
||||
logger.warning(f"解析帖子项异常: {e}")
|
||||
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 _extract_post(self, item, base_url: str) -> Optional[Dict]:
|
||||
post_id = None
|
||||
for attr in ["data-id", "data-post-id", "id"]:
|
||||
val = item.get(attr)
|
||||
if val:
|
||||
post_id = str(val)
|
||||
break
|
||||
if not post_id:
|
||||
return None
|
||||
def parse_post_detail(self, html: str, url: str) -> Optional[Dict]:
|
||||
"""解析帖子详情页"""
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
|
||||
title_elem = item.select_one(".title, .thread-title, .subject")
|
||||
title = title_elem.get_text(strip=True) if title_elem else f"无标题_{post_id}"
|
||||
author_elem = item.select_one(".author, .thread-author, .username")
|
||||
author_text = author_elem.get_text(strip=True) if author_elem else "匿名"
|
||||
author_id = None
|
||||
for attr in ["data-author-id", "data-user-id", "data-uid"]:
|
||||
val = item.get(attr)
|
||||
if val:
|
||||
author_id = str(val)
|
||||
break
|
||||
# 提取帖子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_elem = item.select_one(".price, .deal-price, .cost")
|
||||
if price_elem:
|
||||
price_text = price_elem.get_text(strip=True)
|
||||
match = re.search(r"[\d.]+", price_text.replace(",", ""))
|
||||
if match:
|
||||
price = float(match.group())
|
||||
if "条" in price_text:
|
||||
price_unit = "元/条"
|
||||
elif "张" in price_text:
|
||||
price_unit = "元/张"
|
||||
else:
|
||||
price_unit = "元"
|
||||
|
||||
view_count = 0
|
||||
reply_count = 0
|
||||
like_count = 0
|
||||
view_elem = item.select_one(".views, .view-count")
|
||||
if view_elem:
|
||||
match = re.search(r"[\d]+", view_elem.get_text())
|
||||
if match:
|
||||
view_count = int(match.group())
|
||||
reply_elem = item.select_one(".replies, .reply-count")
|
||||
if reply_elem:
|
||||
match = re.search(r"[\d]+", reply_elem.get_text())
|
||||
if match:
|
||||
reply_count = int(match.group())
|
||||
like_elem = item.select_one(".likes, .like-count")
|
||||
if like_elem:
|
||||
match = re.search(r"[\d]+", like_elem.get_text())
|
||||
if match:
|
||||
like_count = int(match.group())
|
||||
# 优先从标题/内容匹配价格
|
||||
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_elem = item.select_one(".time, .created-at, .post-time")
|
||||
if time_elem:
|
||||
created_at = self._parse_datetime(time_elem.get_text(strip=True))
|
||||
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
|
||||
|
||||
post_type = "normal"
|
||||
class_attr = item.get("class", [])
|
||||
if "deal" in class_attr or "trade" in class_attr:
|
||||
post_type = "deal"
|
||||
|
||||
is_top = False
|
||||
is_essence = False
|
||||
badge_elems = item.select(".badge, .tag")
|
||||
for badge in badge_elems:
|
||||
text = badge.get_text(strip=True).lower()
|
||||
if "顶" in text or "top" in text:
|
||||
is_top = True
|
||||
if "精" in text or "ess" in text:
|
||||
is_essence = True
|
||||
# 提取内容（主要文本）
|
||||
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, "topic_id": post_id, "title": title,
|
||||
"content": None, "content_html": None,
|
||||
"author_id": author_id or f"user_{author_text}", "author_username": author_text,
|
||||
"category": None, "sub_category": None, "post_type": post_type,
|
||||
"price": price, "price_unit": price_unit,
|
||||
"view_count": view_count, "reply_count": reply_count, "like_count": like_count,
|
||||
"is_top": is_top, "is_essence": is_essence, "is_closed": False,
|
||||
"created_at": created_at, "updated_at": created_at,
|
||||
'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 parse_post_detail(self, html: str, url: str) -> Optional[Dict]:
|
||||
soup = BeautifulSoup(html, "lxml")
|
||||
post_id = None
|
||||
match = re.search(r"/thread/(\d+)", url)
|
||||
if match:
|
||||
post_id = match.group(1)
|
||||
title_elem = soup.select_one("h1.title, h1.thread-title, .post-title")
|
||||
title = title_elem.get_text(strip=True) if title_elem else None
|
||||
content_elem = soup.select_one(".post-content, .thread-content, .content")
|
||||
content = content_elem.get_text(strip=True, separator="\n") if content_elem else None
|
||||
return {"post_id": post_id, "topic_id": post_id, "title": title, "content": content}
|
||||
|
||||
def _parse_datetime(self, time_str: str) -> Optional[str]:
|
||||
if not time_str:
|
||||
return None
|
||||
time_str = time_str.strip()
|
||||
def _extract_price(self, text: str) -> Optional[float]:
|
||||
"""从标题提取价格"""
|
||||
patterns = [
|
||||
(r"\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}", "%Y-%m-%d %H:%M:%S"),
|
||||
(r"\d{4}-\d{2}-\d{2}", "%Y-%m-%d"),
|
||||
(r"\d+分钟前", "minutes_ago"),
|
||||
(r"\d+小时前", "hours_ago"),
|
||||
r'(\d+)\s*[万Ww]?\s*元',
|
||||
r'(\d+)\s*-\s*(\d+)\s*[万Ww]?',
|
||||
r'[收售出求兑买]\s*(\d+)',
|
||||
]
|
||||
for pattern, fmt in patterns:
|
||||
match = re.search(pattern, time_str)
|
||||
if match:
|
||||
if fmt == "minutes_ago":
|
||||
mins = int(re.search(r"\d+", match.group()).group())
|
||||
dt = datetime.now() - timedelta(minutes=mins)
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
elif fmt == "hours_ago":
|
||||
hours = int(re.search(r"\d+", match.group()).group())
|
||||
dt = datetime.now() - timedelta(hours=hours)
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
else:
|
||||
try:
|
||||
dt = datetime.strptime(match.group(), fmt)
|
||||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except:
|
||||
pass
|
||||
for pattern in patterns:
|
||||
m = re.search(pattern, text)
|
||||
if m:
|
||||
try:
|
||||
return float(m.group(1))
|
||||
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 save_post(self, post: Dict) -> bool:
|
||||
if not post or not post.get("post_id"):
|
||||
"""保存帖子到数据库"""
|
||||
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 = VALUES(title), content = VALUES(content),
|
||||
view_count = VALUES(view_count), reply_count = VALUES(reply_count),
|
||||
updated_at = VALUES(updated_at), crawled_at = NOW()
|
||||
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")
|
||||
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')
|
||||
))
|
||||
logger.info(f"帖子保存成功: {str(post.get('title'))[:30]}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"保存帖子失败: {e}")
|
||||
return False
|
||||
|
||||
def crawl_forum(self, category_key: str = "longchao", max_pages: int = 5) -> List[Dict]:
|
||||
if category_key not in self.categories:
|
||||
logger.error(f"未知板块: {category_key}")
|
||||
return []
|
||||
category = self.categories[category_key]
|
||||
base_url = f"{self.base_url}{category['url']}"
|
||||
logger.info(f"开始爬取板块: {category['name']} ({base_url})")
|
||||
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"{base_url}?page={page}"
|
||||
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} 页无数据")
|
||||
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"板块 {category['name']} 共采集 {len(all_posts)} 条帖子")
|
||||
|
||||
logger.info(f"连体纪念钞板块共采集 {len(all_posts)} 条帖子")
|
||||
return all_posts
|
||||
|
||||
def run(self, category: str = "longchao") -> List[Dict]:
|
||||
def run(self, max_pages: int = 5, crawl_detail: bool = False) -> List[Dict]:
|
||||
"""执行爬虫"""
|
||||
log_id = self._log_start()
|
||||
|
||||
try:
|
||||
posts = self.crawl_forum(category, self.max_pages)
|
||||
posts = self.crawl_forum(max_pages, crawl_detail)
|
||||
self._log_finish(log_id, "success", len(posts))
|
||||
return posts
|
||||
except Exception as e:
|
||||
|
|
@ -315,20 +351,28 @@ class YichensPostSpider(PaginationSpider):
|
|||
|
||||
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"))
|
||||
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))
|
||||
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(category: str = "longchao") -> List[Dict]:
|
||||
spider = YichensPostSpider()
|
||||
return spider.run(category)
|
||||
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
|
||||
category = sys.argv[1] if len(sys.argv) > 1 else "longchao"
|
||||
crawl_yichens(category)
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,19 @@
|
|||
#!/bin/bash
|
||||
# 每日一尘网爬虫任务
|
||||
DATE=$(date +%Y-%m-%d)
|
||||
LOG_FILE="/root/coolbot-data/logs/crawl_${DATE}.log"
|
||||
|
||||
echo "[$(date)] 开始每日爬虫任务..." >> $LOG_FILE
|
||||
|
||||
cd /root/coolbot-data
|
||||
source venv/bin/activate
|
||||
export PYTHONPATH=/root/coolbot-data:$PYTHONPATH
|
||||
|
||||
python3 -c "
|
||||
from crawlers.yichens_spider import YichensSpider
|
||||
spider = YichensSpider()
|
||||
posts = spider.run(max_pages=5)
|
||||
print(f'采集帖子数: {len(posts)}')
|
||||
" >> $LOG_FILE 2>&1
|
||||
|
||||
echo "[$(date)] 爬虫任务完成" >> $LOG_FILE
|
||||
|
|
@ -0,0 +1,190 @@
|
|||
#!/usr/bin/env python3
|
||||
"""CoolBotDataSys 每日飞书日报推送"""
|
||||
import sys
|
||||
import os
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
import requests
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from database import db
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def get_collection_stats():
|
||||
"""获取藏品统计"""
|
||||
with db.get_cursor() as cursor:
|
||||
cursor.execute("SELECT COUNT(*) as cnt FROM collections")
|
||||
total = cursor.fetchone()["cnt"]
|
||||
return {"total_collections": total}
|
||||
|
||||
def get_yichens_stats():
|
||||
"""获取一尘数据统计"""
|
||||
with db.get_cursor() as cursor:
|
||||
# 总帖子数
|
||||
cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts")
|
||||
total_posts = cursor.fetchone()["cnt"]
|
||||
|
||||
# 今日新增
|
||||
cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE DATE(crawled_at) = CURDATE()")
|
||||
today_posts = cursor.fetchone()["cnt"]
|
||||
|
||||
# 交易帖数量
|
||||
cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE post_type = 'deal'")
|
||||
deal_posts = cursor.fetchone()["cnt"]
|
||||
|
||||
# 有价格标注的帖子
|
||||
cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE price IS NOT NULL AND price > 0")
|
||||
priced_posts = cursor.fetchone()["cnt"]
|
||||
|
||||
# 今日价格区间统计
|
||||
cursor.execute("""
|
||||
SELECT
|
||||
COUNT(*) as cnt,
|
||||
AVG(price) as avg_price,
|
||||
MIN(price) as min_price,
|
||||
MAX(price) as max_price
|
||||
FROM yichens_posts
|
||||
WHERE price IS NOT NULL AND price > 0 AND DATE(crawled_at) = CURDATE()
|
||||
""")
|
||||
today_price_stats = cursor.fetchone()
|
||||
|
||||
# 最新帖子（标题示例）
|
||||
cursor.execute("""
|
||||
SELECT title, price, price_unit
|
||||
FROM yichens_posts
|
||||
WHERE price IS NOT NULL AND price > 0
|
||||
ORDER BY crawled_at DESC LIMIT 5
|
||||
""")
|
||||
latest_with_price = cursor.fetchall()
|
||||
|
||||
# 最近爬虫状态
|
||||
cursor.execute("""
|
||||
SELECT source, status, items_count, finished_at
|
||||
FROM crawl_logs
|
||||
ORDER BY finished_at DESC LIMIT 3
|
||||
""")
|
||||
crawl_status = cursor.fetchall()
|
||||
|
||||
return {
|
||||
"total_posts": total_posts,
|
||||
"today_posts": today_posts,
|
||||
"deal_posts": deal_posts,
|
||||
"priced_posts": priced_posts,
|
||||
"today_price_stats": today_price_stats,
|
||||
"latest_with_price": latest_with_price,
|
||||
"crawl_status": crawl_status
|
||||
}
|
||||
|
||||
def build_feishu_message(collection_stats, yichens_stats):
|
||||
"""构建飞书消息卡片"""
|
||||
now = datetime.now()
|
||||
|
||||
# 最新价格帖子示例
|
||||
latest_items = ""
|
||||
for item in yichens_stats.get("latest_with_price", []):
|
||||
price_str = f"¥{item['price']}{item['price_unit']}" if item['price'] else "价格待确认"
|
||||
latest_items += f"- {item['title'][:30]}... : {price_str}\n"
|
||||
if not latest_items:
|
||||
latest_items = "今日暂无价格数据\n"
|
||||
|
||||
# 爬虫状态
|
||||
crawl_info = ""
|
||||
for log in yichens_stats.get("crawl_status", []):
|
||||
status_emoji = "✅" if log["status"] == "success" else "❌"
|
||||
crawl_info += f"{status_emoji} {log['source']}: {log['items_count']}条 @ {log['finished_at']}\n"
|
||||
if not crawl_info:
|
||||
crawl_info = "暂无爬虫运行记录\n"
|
||||
|
||||
# 价格统计
|
||||
price_stats = yichens_stats.get("today_price_stats", {})
|
||||
price_info = ""
|
||||
if price_stats and price_stats.get("cnt", 0) > 0:
|
||||
price_info = f"均价: ¥{price_stats['avg_price']:.0f} | 区间: ¥{price_stats['min_price']:.0f}~¥{price_stats['max_price']:.0f} ({price_stats['cnt']}条)"
|
||||
else:
|
||||
price_info = "今日暂无价格统计"
|
||||
|
||||
message = {
|
||||
"msg_type": "interactive",
|
||||
"card": {
|
||||
"header": {
|
||||
"title": {"tag": "plain_text", "content": f"📊 CoolBotDataSys 日报 {now.strftime('%Y-%m-%d %H:%M')}"},
|
||||
"template": "blue"
|
||||
},
|
||||
"elements": [
|
||||
{
|
||||
"tag": "div",
|
||||
"text": {
|
||||
"tag": "lark_md",
|
||||
"content": (
|
||||
f"**🐉 一尘连体纪念钞数据**\n"
|
||||
f"- 总帖子: {yichens_stats['total_posts']} 条\n"
|
||||
f"- 今日新增: +{yichens_stats['today_posts']} 条\n"
|
||||
f"- 交易帖: {yichens_stats['deal_posts']} 条\n"
|
||||
f"- 有价格: {yichens_stats['priced_posts']} 条\n\n"
|
||||
f"**💰 今日价格动态**\n{price_info}\n\n"
|
||||
f"**📈 最新交易帖**\n{latest_items}\n\n"
|
||||
f"**🔄 爬虫状态**\n{crawl_info}"
|
||||
)
|
||||
}
|
||||
},
|
||||
{"tag": "hr"},
|
||||
{
|
||||
"tag": "note",
|
||||
"elements": [
|
||||
{"tag": "plain_text", "content": f"CoolBotDataSys · {now.strftime('%H:%M:%S')}"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
return message
|
||||
|
||||
def send_feishu(message):
|
||||
"""发送飞书消息"""
|
||||
webhook = os.environ.get("FEISHU_WEBHOOK_URL", "")
|
||||
if not webhook:
|
||||
logger.warning("飞书Webhook未配置，跳过发送")
|
||||
return False
|
||||
|
||||
try:
|
||||
response = requests.post(webhook, json=message, timeout=10)
|
||||
result = response.json()
|
||||
if result.get("code") == 0 or result.get("StatusCode") == 0:
|
||||
logger.info("✅ 飞书消息发送成功")
|
||||
return True
|
||||
else:
|
||||
logger.error(f"❌ 飞书发送失败: {result}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 飞书发送异常: {e}")
|
||||
return False
|
||||
|
||||
def main():
|
||||
logger.info("📊 开始生成日报...")
|
||||
|
||||
collection_stats = get_collection_stats()
|
||||
yichens_stats = get_yichens_stats()
|
||||
|
||||
print(f"统计概览:")
|
||||
print(f" 一尘帖子总数: {yichens_stats['total_posts']}")
|
||||
print(f" 今日新增: {yichens_stats['today_posts']}")
|
||||
print(f" 交易帖: {yichens_stats['deal_posts']}")
|
||||
print(f" 有价格标注: {yichens_stats['priced_posts']}")
|
||||
|
||||
message = build_feishu_message(collection_stats, yichens_stats)
|
||||
print(f"\n消息卡片内容预览:")
|
||||
print(message["card"]["elements"][0]["text"]["content"][:500])
|
||||
|
||||
# 检查是否启用飞书推送
|
||||
webhook = os.environ.get("FEISHU_WEBHOOK_URL", "")
|
||||
if webhook:
|
||||
send_feishu(message)
|
||||
else:
|
||||
print("\n⚠️ 未配置 FEISHU_WEBHOOK_URL 环境变量，跳过飞书推送")
|
||||
|
||||
return yichens_stats
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
#!/bin/bash
|
||||
cd /root/coolbot-data
|
||||
source venv/bin/activate
|
||||
export PYTHONPATH=/root/coolbot-data:$PYTHONPATH
|
||||
python3 scripts/daily_report.py >> logs/report.log 2>&1
|
||||
Loading…
Reference in New Issue