import json import csv import os import re import time import requests from datetime import datetime from itemadapter import ItemAdapter class JsonPipeline: """保存爬取结果到JSON文件""" def __init__(self): self.success_file = None self.fail_file = None self.success_path = None def open_spider(self, spider): os.makedirs('output', exist_ok=True) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') self.success_path = f'output/{spider.name}_success_{timestamp}.jsonl' self.fail_path = f'output/{spider.name}_failed_{timestamp}.jsonl' self.success_file = open(self.success_path, 'w', encoding='utf-8') self.fail_file = open(self.fail_path, 'w', encoding='utf-8') def close_spider(self, spider): self.success_file.close() self.fail_file.close() spider.logger.info(f'爬取完成: {self.success_path}') # 自动推送到搜索服务器 self._auto_push(spider) def _auto_push(self, spider): """自动将爬取数据推送到华为云搜索库""" import subprocess import os key = os.path.expanduser("~/Downloads/KeyPair-a00a.pem") server = "root@1.94.217.116" local_file = self.success_path if not os.path.exists(local_file): spider.logger.warning('推送失败: 文件不存在') return size = os.path.getsize(local_file) if size == 0: spider.logger.warning('推送跳过: 无新数据') return spider.logger.info(f'自动推送到搜索库 ({size//1024}KB)...') try: # 压缩传输 import gzip with open(local_file, 'rb') as f: data = gzip.compress(f.read()) remote_name = os.path.basename(local_file) + '.gz' # scp via subprocess import tempfile tmp = os.path.join(tempfile.gettempdir(), remote_name) with open(tmp, 'wb') as f: f.write(data) # 传输 subprocess.run( ['scp', '-i', key, '-o', 'StrictHostKeyChecking=no', tmp, f'{server}:/tmp/{remote_name}'], capture_output=True, timeout=120 ) # 服务器上解压导入 result = subprocess.run( ['ssh', '-i', key, '-o', 'StrictHostKeyChecking=no', server, f'cd /root && gunzip -f /tmp/{remote_name} && python3 import_data.py /tmp/{os.path.basename(local_file)} 2>&1 && rm -f /tmp/{os.path.basename(local_file)}'], capture_output=True, text=True, timeout=300 ) os.remove(tmp) if result.returncode == 0: # 提取导入数量 for line in result.stdout.split('\n'): if 'imported' in line or 'already' in line: spider.logger.info(f'推送结果: {line.strip()}') spider.logger.info('✅ 数据已同步到搜索库') else: spider.logger.error(f'推送失败: {result.stderr[:200]}') except Exception as e: spider.logger.error(f'推送异常: {e}') def process_item(self, item, spider): adapter = ItemAdapter(item) line = json.dumps(dict(adapter), ensure_ascii=False) + '\n' if adapter.get('status') == 'failed': self.fail_file.write(line) else: self.success_file.write(line) return item class CsvPipeline: """也存一份CSV方便查看""" def __init__(self): self.writer = None self.file = None def open_spider(self, spider): os.makedirs('output', exist_ok=True) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') self.file = open(f'output/{spider.name}_result_{timestamp}.csv', 'w', encoding='utf-8-sig', newline='') self.writer = csv.writer(self.file) self.writer.writerow(['站点名', '原始URL', '页面URL', '标题', '发布日期', '正文摘要', '状态', '备注']) def close_spider(self, spider): self.file.close() def process_item(self, item, spider): adapter = ItemAdapter(item) content = adapter.get('content', '') or '' self.writer.writerow([ adapter.get('site_name', ''), adapter.get('source_url', ''), adapter.get('page_url', ''), adapter.get('title', ''), adapter.get('publish_date', ''), content[:2000] if content else '', adapter.get('status', ''), adapter.get('error_msg', ''), ]) return item class NotionKeywordPipeline: """关键词命中 → 推送到 Notion 数据库 配置（在 settings.py 中设置）： - NOTION_API_KEY: Notion Integration Token - NOTION_DATABASE_ID: 目标数据库 ID - NOTION_KEYWORDS: 关键词列表 ["化工", "乙烯", ...] - NOTION_PUSH_INTERVAL: 推送间隔秒数（防限流，默认 0.5） """ def __init__(self, api_key, database_id, keywords, interval): self.api_key = api_key self.database_id = database_id self.keywords = keywords self.interval = interval self.seen_urls = set() self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Notion-Version": "2022-06-28", } self.last_push = 0.0 self.push_count = 0 @classmethod def from_crawler(cls, crawler): return cls( api_key=crawler.settings.get("NOTION_API_KEY"), database_id=crawler.settings.get("NOTION_DATABASE_ID"), keywords=crawler.settings.get("NOTION_KEYWORDS", []), interval=crawler.settings.getfloat("NOTION_PUSH_INTERVAL", 0.5), ) def _matches_keywords(self, text): """检查文本是否命中任一关键词""" if not text: return False text_lower = text.lower() for kw in self.keywords: if kw.lower() in text_lower: return True return False def _push_to_notion(self, item, spider): """推送到 Notion 数据库页面""" adapter = ItemAdapter(item) url = adapter.get("page_url") or adapter.get("source_url", "") # 去重 if url in self.seen_urls: return False self.seen_urls.add(url) # 速率限制 elapsed = time.time() - self.last_push if elapsed < self.interval: time.sleep(self.interval - elapsed) title = (adapter.get("title") or "")[:100] content = (adapter.get("content") or "")[:3000] summary = re.sub(r'<[^>]+>', '', content).strip()[:500] date_str = adapter.get("publish_date") or "" data = { "parent": {"database_id": self.database_id}, "properties": { "标题": {"title": [{"text": {"content": title}}]}, "站点": {"select": {"name": adapter.get("site_name", "")}}, "日期": {"date": {"start": date_str}} if date_str else {"rich_text": [{"text": {"content": ""}}]}, "链接": {"url": url}, "内容摘要": {"rich_text": [{"text": {"content": summary}}]}, "来源域名": {"rich_text": [{"text": {"content": adapter.get("domain", "")}}]}, }, } resp = requests.post( "https://api.notion.com/v1/pages", json=data, headers=self.headers, timeout=15 ) self.last_push = time.time() if resp.status_code == 200: self.push_count += 1 spider.logger.info(f" 📤 Notion推送成功 [{self.push_count}]: {title[:40]}") return True else: spider.logger.warning(f" ⚠️ Notion推送失败 ({resp.status_code}): {title[:40]}") return False def process_item(self, item, spider): adapter = ItemAdapter(item) text = (adapter.get("title") or "") + " " + (adapter.get("content") or "") if self._matches_keywords(text): self._push_to_notion(item, spider) return item def close_spider(self, spider): if self.push_count > 0: spider.logger.info(f"📊 Notion推送总计: {self.push_count} 条")