From f1d46898ed11cd2bd4d8c0af035797950cd04018 Mon Sep 17 00:00:00 2001 From: li Date: Mon, 30 Mar 2026 15:15:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E9=97=BB=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E9=87=87=E9=9B=86=E8=84=9A=E6=9C=AC=20-=20CoinDesk/CryptoCompa?= =?UTF-8?q?re=E5=8F=8C=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 主源CryptoCompare API + 备源CoinDesk RSS - MD5去重, 30天旧新闻自动清理 - 宝塔计划任务每6小时执行 --- crypto_news_crawler.py | 247 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 crypto_news_crawler.py diff --git a/crypto_news_crawler.py b/crypto_news_crawler.py new file mode 100644 index 0000000..92eb170 --- /dev/null +++ b/crypto_news_crawler.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Crypto News Crawler for BT Panel Scheduled Task +数据源: CryptoCompare News API (免费) + CoinDesk RSS (备用) +目标表: fa_app_zixun +""" + +import requests +import pymysql +import time +import hashlib +import re +import sys +import logging +from datetime import datetime + +# ============ 数据库配置 - 根据实际服务器修改 ============ +DB_CONFIG = { + 'host': '127.0.0.1', + 'port': 3306, + 'user': 'jys', + 'password': 'YmJKJ6JcRXK7AbzT', + 'database': 'jys', + 'charset': 'utf8mb4', +} + +TABLE_PREFIX = 'fa_' +TABLE_NAME = f'{TABLE_PREFIX}app_zixun' +ADMIN_USER_ID = 1 # 后台管理员 user_id +MAX_NEWS_PER_RUN = 20 # 每次最多采集条数 + +# 日志配置 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s', + handlers=[ + logging.StreamHandler(sys.stdout), + ] +) +log = logging.getLogger(__name__) + + +def get_db_connection(): + """获取数据库连接""" + return pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.DictCursor) + + +def news_exists(cursor, title_hash): + """检查新闻是否已存在（通过标题哈希去重）""" + sql = f"SELECT id FROM {TABLE_NAME} WHERE title_hash = %s LIMIT 1" + cursor.execute(sql, (title_hash,)) + return cursor.fetchone() is not None + + +def ensure_title_hash_column(cursor): + """确保 title_hash 列存在（用于去重）""" + cursor.execute(f"SHOW COLUMNS FROM {TABLE_NAME} LIKE 'title_hash'") + if not cursor.fetchone(): + log.info("Adding title_hash column for deduplication...") + cursor.execute(f"ALTER TABLE {TABLE_NAME} ADD COLUMN title_hash VARCHAR(32) DEFAULT '' AFTER content_en") + cursor.execute(f"CREATE INDEX idx_title_hash ON {TABLE_NAME}(title_hash)") + log.info("title_hash column added.") + + +def clean_html(text): + """清除 HTML 标签""" + if not text: + return '' + clean = re.sub(r'<[^>]+>', '', text) + clean = re.sub(r'&[a-zA-Z]+;', ' ', clean) + clean = re.sub(r'\s+', ' ', clean).strip() + return clean + + +def fetch_cryptocompare_news(): + """从 CryptoCompare 获取加密货币新闻""" + url = "https://min-api.cryptocompare.com/data/v2/news/?lang=EN&sortOrder=latest" + headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' + } + try: + resp = requests.get(url, headers=headers, timeout=30) + resp.raise_for_status() + data = resp.json() + if data.get('Type') == 100 and data.get('Data'): + news_list = [] + for item in data['Data'][:MAX_NEWS_PER_RUN]: + news_list.append({ + 'title_en': item.get('title', ''), + 'content_en': clean_html(item.get('body', '')), + 'cover_image': item.get('imageurl', ''), + 'published_on': item.get('published_on', int(time.time())), + 'source': item.get('source', 'CryptoCompare'), + 'url': item.get('url', ''), + }) + log.info(f"CryptoCompare: fetched {len(news_list)} articles") + return news_list + except Exception as e: + log.error(f"CryptoCompare fetch failed: {e}") + return [] + + +def fetch_coindesk_rss(): + """从 CoinDesk RSS 获取新闻（备用）""" + url = "https://www.coindesk.com/arc/outboundfeeds/rss/" + headers = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' + } + try: + resp = requests.get(url, headers=headers, timeout=30) + resp.raise_for_status() + # 简单 XML 解析 RSS + import xml.etree.ElementTree as ET + root = ET.fromstring(resp.content) + news_list = [] + for item in root.findall('.//item')[:MAX_NEWS_PER_RUN]: + title = item.findtext('title', '') + description = clean_html(item.findtext('description', '')) + pub_date = item.findtext('pubDate', '') + # 解析 RSS 日期 + try: + dt = datetime.strptime(pub_date, '%a, %d %b %Y %H:%M:%S %z') + timestamp = int(dt.timestamp()) + except: + timestamp = int(time.time()) + + news_list.append({ + 'title_en': title, + 'content_en': description, + 'cover_image': '', + 'published_on': timestamp, + 'source': 'CoinDesk', + 'url': item.findtext('link', ''), + }) + log.info(f"CoinDesk RSS: fetched {len(news_list)} articles") + return news_list + except Exception as e: + log.error(f"CoinDesk RSS fetch failed: {e}") + return [] + + +def insert_news(news_list): + """将新闻插入数据库""" + if not news_list: + log.info("No news to insert.") + return 0 + + conn = get_db_connection() + cursor = conn.cursor() + inserted = 0 + + try: + ensure_title_hash_column(cursor) + conn.commit() + + for news in news_list: + title_en = news['title_en'][:500] if news['title_en'] else '' + if not title_en: + continue + + title_hash = hashlib.md5(title_en.encode('utf-8')).hexdigest() + + if news_exists(cursor, title_hash): + continue + + # title 和 content 使用英文（此项目面向英文用户） + sql = f"""INSERT INTO {TABLE_NAME} + (user_id, title, title_en, content, content_en, cover_image, + addtime, look_num, title_hash) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""" + + content_en = news['content_en'][:5000] if news['content_en'] else title_en + + cursor.execute(sql, ( + ADMIN_USER_ID, + title_en, # title (中文字段也填英文) + title_en, # title_en + content_en, # content + content_en, # content_en + news.get('cover_image', ''), + news['published_on'], + 0, # look_num + title_hash, + )) + inserted += 1 + + conn.commit() + log.info(f"Inserted {inserted} new articles into database.") + + except Exception as e: + conn.rollback() + log.error(f"Database insert failed: {e}") + finally: + cursor.close() + conn.close() + + return inserted + + +def cleanup_old_news(days=30): + """清理超过 N 天的旧新闻""" + conn = get_db_connection() + cursor = conn.cursor() + try: + cutoff = int(time.time()) - (days * 86400) + sql = f"DELETE FROM {TABLE_NAME} WHERE addtime < %s AND addtime > 0" + cursor.execute(sql, (cutoff,)) + deleted = cursor.rowcount + conn.commit() + if deleted > 0: + log.info(f"Cleaned up {deleted} old articles (older than {days} days).") + except Exception as e: + conn.rollback() + log.error(f"Cleanup failed: {e}") + finally: + cursor.close() + conn.close() + + +def main(): + log.info("=" * 50) + log.info("Crypto News Crawler started") + log.info("=" * 50) + + # 主数据源: CryptoCompare + news = fetch_cryptocompare_news() + + # 备用数据源: CoinDesk RSS + if not news: + log.info("Primary source failed, trying CoinDesk RSS...") + news = fetch_coindesk_rss() + + if news: + count = insert_news(news) + log.info(f"Total new articles: {count}") + else: + log.warning("All sources failed. No news fetched.") + + # 清理30天前的旧新闻 + cleanup_old_news(30) + + log.info("Crawler finished.") + + +if __name__ == '__main__': + main()