diff --git a/api/main.py b/api/main.py index 199d993..39aae5c 100644 --- a/api/main.py +++ b/api/main.py @@ -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, diff --git a/crawlers/yichens_spider.py b/crawlers/yichens_spider.py index a322128..7b84498 100644 --- a/crawlers/yichens_spider.py +++ b/crawlers/yichens_spider.py @@ -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']*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 _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 = 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) diff --git a/scripts/daily_crawl.sh b/scripts/daily_crawl.sh new file mode 100755 index 0000000..8a56437 --- /dev/null +++ b/scripts/daily_crawl.sh @@ -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 diff --git a/scripts/daily_report.py b/scripts/daily_report.py new file mode 100644 index 0000000..c3659b1 --- /dev/null +++ b/scripts/daily_report.py @@ -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() diff --git a/scripts/daily_report.sh b/scripts/daily_report.sh new file mode 100755 index 0000000..7f5a89c --- /dev/null +++ b/scripts/daily_report.sh @@ -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