Daily Paper Digest

{
  "ownerId": "kn791b3prmq172nkkkf52kg8y181ep3e",
  "slug": "daily-paper-digest",
  "version": "1.0.0",
  "publishedAt": 1771587246433
}

{
  "version": 1,
  "registry": "https://clawhub.ai",
  "slug": "daily-paper-digest",
  "installedVersion": "1.0.0",
  "installedAt": 1776068341793
}

"""
arXiv 论文爬取模块
"""
import arxiv
from datetime import datetime, timedelta
from typing import List, Dict
import logging

logger = logging.getLogger(__name__)


class ArxivFetcher:
    """arXiv 论文获取器"""
    
    def __init__(self, categories: List[str], max_results: int = 10):
        """
        初始化 arXiv 获取器
        
        Args:
            categories: 论文分类列表，如 ['cs.AI', 'cs.CL']
            max_results: 每个分类最大返回结果数
        """
        self.categories = categories
        self.max_results = max_results
    
    def fetch_daily_papers(self) -> List[Dict]:
        """
        获取每日最新论文
        
        Returns:
            论文列表，每篇论文包含标题、作者、摘要、链接等信息
        """
        papers = []
        yesterday = datetime.now() - timedelta(days=1)
        date_str = yesterday.strftime('%Y%m%d')
        
        for category in self.categories:
            try:
                logger.info(f"正在获取 {category} 分类的论文...")
                
                # 构建查询
                search = arxiv.Search(
                    query=f"cat:{category}",
                    max_results=self.max_results,
                    sort_by=arxiv.SortCriterion.SubmittedDate,
                    sort_order=arxiv.SortOrder.Descending
                )
                
                # 获取结果
                for result in search.results():
                    # 只获取最近一天的论文
                    if result.published.date() >= yesterday.date():
                        paper = {
                            'title': result.title,
                            'authors': [author.name for author in result.authors],
                            'abstract': result.summary,
                            'pdf_url': result.pdf_url,
                            'arxiv_url': result.entry_id,
                            'published': result.published.strftime('%Y-%m-%d'),
                            'category': category,
                            'source': 'arxiv'
                        }
                        papers.append(paper)
                        
            except Exception as e:
                logger.error(f"获取 {category} 论文失败: {str(e)}")
        
        logger.info(f"从 arXiv 获取了 {len(papers)} 篇论文")
        return papers
    
    def search_papers(self, query: str, max_results: int = 5) -> List[Dict]:
        """
        搜索特定主题的论文
        
        Args:
            query: 搜索关键词
            max_results: 最大返回结果数
            
        Returns:
            论文列表
        """
        papers = []
        
        try:
            search = arxiv.Search(
                query=query,
                max_results=max_results,
                sort_by=arxiv.SortCriterion.SubmittedDate,
                sort_order=arxiv.SortOrder.Descending
            )
            
            for result in search.results():
                paper = {
                    'title': result.title,
                    'authors': [author.name for author in result.authors],
                    'abstract': result.summary,
                    'pdf_url': result.pdf_url,
                    'arxiv_url': result.entry_id,
                    'published': result.published.strftime('%Y-%m-%d'),
                    'source': 'arxiv'
                }
                papers.append(paper)
                
        except Exception as e:
            logger.error(f"搜索论文失败: {str(e)}")
        
        return papers


if __name__ == "__main__":
    # 测试代码
    logging.basicConfig(level=logging.INFO)
    fetcher = ArxivFetcher(categories=['cs.AI', 'cs.CL'], max_results=5)
    papers = fetcher.fetch_daily_papers()
    
    print(f"\n找到 {len(papers)} 篇论文：")
    for i, paper in enumerate(papers, 1):
        print(f"\n{i}. {paper['title']}")
        print(f"   作者: {', '.join(paper['authors'][:3])}")
        print(f"   链接: {paper['arxiv_url']}")

{
  "sources": [
    {
      "name": "arxiv",
      "enabled": true,
      "description": "arXiv每日论文速递 - 专注于大语言模型",
      "categories": [
        "cs.CL",
        "cs.AI",
        "cs.LG"
      ],
      "max_results": 5,
      "update_interval": "daily"
    },
    {
      "name": "huggingface",
      "enabled": true,
      "description": "HuggingFace每日热门论文",
      "url": "https://huggingface.co/papers",
      "max_results": 5,
      "update_interval": "daily"
    }
  ],
  "output_format": {
    "include_abstract": true,
    "include_authors": true,
    "include_links": true,
    "language": "zh-CN"
  },
  "filter": {
    "keywords": [
      "LLM",
      "Large Language Model",
      "GPT",
      "Transformer",
      "BERT",
      "ChatGPT",
      "agent",
      "reasoning"
    ],
    "exclude_keywords": []
  }
}

{
  "sources": [
    {
      "name": "arxiv",
      "enabled": true,
      "description": "arXiv每日论文速递",
      "categories": [
        "cs.AI",
        "cs.CL",
        "cs.CV",
        "cs.LG",
        "cs.NE"
      ],
      "max_results": 10,
      "update_interval": "daily"
    },
    {
      "name": "huggingface",
      "enabled": true,
      "description": "HuggingFace每日论文",
      "url": "https://huggingface.co/papers",
      "max_results": 10,
      "update_interval": "daily"
    }
  ],
  "output_format": {
    "include_abstract": true,
    "include_authors": true,
    "include_links": true,
    "language": "zh-CN"
  },
  "filter": {
    "keywords": [],
    "exclude_keywords": []
  }
}

"""
HuggingFace 论文爬取模块
"""
import requests
from bs4 import BeautifulSoup
from typing import List, Dict
import logging
from datetime import datetime

logger = logging.getLogger(__name__)


class HuggingFaceFetcher:
    """HuggingFace 每日论文获取器"""
    
    def __init__(self, max_results: int = 10):
        """
        初始化 HuggingFace 获取器
        
        Args:
            max_results: 最大返回结果数
        """
        self.base_url = "https://huggingface.co/papers"
        self.max_results = max_results
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    
    def fetch_daily_papers(self) -> List[Dict]:
        """
        获取 HuggingFace 每日热门论文
        
        Returns:
            论文列表
        """
        papers = []
        
        try:
            logger.info("正在获取 HuggingFace 每日论文...")
            
            # 请求页面
            response = requests.get(self.base_url, headers=self.headers, timeout=10)
            response.raise_for_status()
            
            # 解析 HTML
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # 查找论文卡片
            paper_cards = soup.find_all('article', class_='overview-card-wrapper', limit=self.max_results)
            
            for card in paper_cards:
                try:
                    # 提取论文信息
                    title_elem = card.find('h3')
                    title = title_elem.text.strip() if title_elem else "未知标题"
                    
                    # 提取链接
                    link_elem = card.find('a', href=True)
                    paper_link = f"https://huggingface.co{link_elem['href']}" if link_elem else ""
                    
                    # 提取作者
                    authors_elem = card.find('p', class_='text-sm')
                    authors = [authors_elem.text.strip()] if authors_elem else []
                    
                    # 提取摘要
                    abstract_elem = card.find('p', class_='line-clamp-3')
                    abstract = abstract_elem.text.strip() if abstract_elem else ""
                    
                    # 提取点赞数
                    likes_elem = card.find('span', class_='text-sm')
                    likes = likes_elem.text.strip() if likes_elem else "0"
                    
                    paper = {
                        'title': title,
                        'authors': authors,
                        'abstract': abstract,
                        'url': paper_link,
                        'likes': likes,
                        'published': datetime.now().strftime('%Y-%m-%d'),
                        'source': 'huggingface'
                    }
                    papers.append(paper)
                    
                except Exception as e:
                    logger.warning(f"解析单篇论文失败: {str(e)}")
                    continue
            
            logger.info(f"从 HuggingFace 获取了 {len(papers)} 篇论文")
            
        except Exception as e:
            logger.error(f"获取 HuggingFace 论文失败: {str(e)}")
        
        return papers
    
    def fetch_paper_details(self, paper_url: str) -> Dict:
        """
        获取单篇论文的详细信息
        
        Args:
            paper_url: 论文 URL
            
        Returns:
            论文详细信息
        """
        try:
            response = requests.get(paper_url, headers=self.headers, timeout=10)
            response.raise_for_status()
            
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # 这里可以根据实际页面结构提取更多信息
            # 返回基本信息
            return {
                'url': paper_url,
                'fetched_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
            
        except Exception as e:
            logger.error(f"获取论文详情失败: {str(e)}")
            return {}


if __name__ == "__main__":
    # 测试代码
    logging.basicConfig(level=logging.INFO)
    fetcher = HuggingFaceFetcher(max_results=5)
    papers = fetcher.fetch_daily_papers()
    
    print(f"\n找到 {len(papers)} 篇论文：")
    for i, paper in enumerate(papers, 1):
        print(f"\n{i}. {paper['title']}")
        print(f"   点赞: {paper['likes']}")
        print(f"   链接: {paper['url']}")

"""
每日论文速递主程序
整合 arXiv 和 HuggingFace 的论文信息
"""
import json
import os
import logging
from typing import List, Dict
from datetime import datetime
from arxiv_fetcher import ArxivFetcher
from huggingface_fetcher import HuggingFaceFetcher

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class PaperDigest:
    """论文速递主类"""
    
    def __init__(self, config_path: str = "config/sources.json"):
        """
        初始化论文速递
        
        Args:
            config_path: 配置文件路径
        """
        self.config = self._load_config(config_path)
        self.arxiv_fetcher = None
        self.hf_fetcher = None
        self._init_fetchers()
    
    def _load_config(self, config_path: str) -> Dict:
        """加载配置文件"""
        try:
            # 获取脚本所在目录
            script_dir = os.path.dirname(os.path.abspath(__file__))
            full_path = os.path.join(script_dir, config_path)
            
            with open(full_path, 'r', encoding='utf-8') as f:
                config = json.load(f)
            logger.info(f"成功加载配置文件: {config_path}")
            return config
        except Exception as e:
            logger.error(f"加载配置文件失败: {str(e)}")
            # 返回默认配置
            return {
                "sources": [],
                "output_format": {
                    "include_abstract": True,
                    "include_authors": True,
                    "include_links": True,
                    "language": "zh-CN"
                }
            }
    
    def _init_fetchers(self):
        """初始化各个信息源的获取器"""
        for source in self.config.get('sources', []):
            if not source.get('enabled', False):
                continue
                
            if source['name'] == 'arxiv':
                categories = source.get('categories', ['cs.AI'])
                max_results = source.get('max_results', 10)
                self.arxiv_fetcher = ArxivFetcher(categories, max_results)
                logger.info("已初始化 arXiv 获取器")
                
            elif source['name'] == 'huggingface':
                max_results = source.get('max_results', 10)
                self.hf_fetcher = HuggingFaceFetcher(max_results)
                logger.info("已初始化 HuggingFace 获取器")
    
    def fetch_all_papers(self) -> List[Dict]:
        """
        从所有启用的信息源获取论文
        
        Returns:
            所有论文的列表
        """
        all_papers = []
        
        # 获取 arXiv 论文
        if self.arxiv_fetcher:
            try:
                arxiv_papers = self.arxiv_fetcher.fetch_daily_papers()
                all_papers.extend(arxiv_papers)
                logger.info(f"从 arXiv 获取了 {len(arxiv_papers)} 篇论文")
            except Exception as e:
                logger.error(f"获取 arXiv 论文失败: {str(e)}")
        
        # 获取 HuggingFace 论文
        if self.hf_fetcher:
            try:
                hf_papers = self.hf_fetcher.fetch_daily_papers()
                all_papers.extend(hf_papers)
                logger.info(f"从 HuggingFace 获取了 {len(hf_papers)} 篇论文")
            except Exception as e:
                logger.error(f"获取 HuggingFace 论文失败: {str(e)}")
        
        return all_papers
    
    def filter_papers(self, papers: List[Dict]) -> List[Dict]:
        """
        根据配置过滤论文
        
        Args:
            papers: 论文列表
            
        Returns:
            过滤后的论文列表
        """
        filter_config = self.config.get('filter', {})
        keywords = filter_config.get('keywords', [])
        exclude_keywords = filter_config.get('exclude_keywords', [])
        
        if not keywords and not exclude_keywords:
            return papers
        
        filtered_papers = []
        
        for paper in papers:
            title = paper.get('title', '').lower()
            abstract = paper.get('abstract', '').lower()
            text = f"{title} {abstract}"
            
            # 检查排除关键词
            if exclude_keywords:
                if any(keyword.lower() in text for keyword in exclude_keywords):
                    continue
            
            # 检查包含关键词
            if keywords:
                if any(keyword.lower() in text for keyword in keywords):
                    filtered_papers.append(paper)
            else:
                filtered_papers.append(paper)
        
        logger.info(f"过滤后剩余 {len(filtered_papers)} 篇论文")
        return filtered_papers
    
    def format_paper(self, paper: Dict, index: int) -> str:
        """
        格式化单篇论文信息
        
        Args:
            paper: 论文信息
            index: 序号
            
        Returns:
            格式化后的文本
        """
        output_format = self.config.get('output_format', {})
        
        # 构建输出
        lines = []
        lines.append(f"\n{'='*60}")
        lines.append(f"📄 论文 {index}")
        lines.append(f"{'='*60}")
        lines.append(f"\n📌 标题: {paper.get('title', '未知')}")
        
        # 作者信息
        if output_format.get('include_authors', True):
            authors = paper.get('authors', [])
            if authors:
                author_str = ', '.join(authors[:5])  # 最多显示5个作者
                if len(authors) > 5:
                    author_str += f" 等 {len(authors)} 人"
                lines.append(f"👥 作者: {author_str}")
        
        # 来源和发布日期
        source = paper.get('source', '未知').upper()
        published = paper.get('published', '未知')
        lines.append(f"🏷️  来源: {source} | 日期: {published}")
        
        # 摘要
        if output_format.get('include_abstract', True):
            abstract = paper.get('abstract', '')
            if abstract:
                # 限制摘要长度
                if len(abstract) > 300:
                    abstract = abstract[:300] + '...'
                lines.append(f"\n📝 摘要:\n{abstract}")
        
        # 链接
        if output_format.get('include_links', True):
            if paper.get('source') == 'arxiv':
                lines.append(f"\n🔗 arXiv: {paper.get('arxiv_url', '')}")
                lines.append(f"📥 PDF: {paper.get('pdf_url', '')}")
            elif paper.get('source') == 'huggingface':
                lines.append(f"\n🔗 链接: {paper.get('url', '')}")
                likes = paper.get('likes', '0')
                lines.append(f"👍 点赞: {likes}")
        
        return '\n'.join(lines)
    
    def format_digest(self, papers: List[Dict]) -> str:
        """
        格式化论文速递报告
        
        Args:
            papers: 论文列表
            
        Returns:
            完整的报告文本
        """
        if not papers:
            return "📭 今日暂无新论文"
        
        # 标题
        title = f"""
╔══════════════════════════════════════════════════════════╗
║           🎓 AI 论文每日速递 - {datetime.now().strftime('%Y年%m月%d日')}           ║
╚══════════════════════════════════════════════════════════╝

📊 今日共收录 {len(papers)} 篇论文

"""
        
        # 格式化每篇论文
        formatted_papers = [self.format_paper(paper, i+1) for i, paper in enumerate(papers)]
        
        # 统计信息
        arxiv_count = sum(1 for p in papers if p.get('source') == 'arxiv')
        hf_count = sum(1 for p in papers if p.get('source') == 'huggingface')
        
        footer = f"""
\n{'='*60}
📈 信息源统计:
   • arXiv: {arxiv_count} 篇
   • HuggingFace: {hf_count} 篇

⏰ 更新时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
{'='*60}
"""
        
        return title + '\n'.join(formatted_papers) + footer
    
    def run(self) -> str:
        """
        执行论文速递
        
        Returns:
            格式化后的报告文本
        """
        logger.info("开始执行每日论文速递...")
        
        # 获取所有论文
        papers = self.fetch_all_papers()
        
        # 过滤论文
        filtered_papers = self.filter_papers(papers)
        
        # 格式化输出
        digest = self.format_digest(filtered_papers)
        
        logger.info("论文速递执行完成")
        return digest


def main():
    """主函数，用于 OpenClaw skill 调用"""
    digest = PaperDigest()
    result = digest.run()
    print(result)
    return result


if __name__ == "__main__":
    main()

# 每日论文速递 Skill 依赖

# arXiv API 客户端
arxiv==2.1.0

# HTTP 请求
requests==2.31.0

# HTML 解析
beautifulsoup4==4.12.3

# XML/Feed 解析
feedparser==6.0.11

# HTML 标签处理
lxml==5.1.0

# 字符编码检测
chardet==5.2.0

# 日期时间处理
python-dateutil==2.8.2

{
  "name": "daily_paper_digest",
  "version": "1.0.0",
  "description": "每日AI论文速递，聚合arxiv和HuggingFace的最新论文",
  "author": "OpenClaw User",
  "type": "scheduled",
  "schedule": "0 9 * * *",
  "trigger": {
    "keywords": ["论文速递", "今日论文", "最新论文", "paper digest"],
    "commands": ["/papers", "/digest"]
  },
  "dependencies": [
    "arxiv",
    "requests",
    "beautifulsoup4",
    "feedparser"
  ],
  "config": {
    "config_file": "config/sources.json"
  },
  "entry": "main.py"
}

"""
测试脚本 - 用于验证各个模块是否正常工作
"""
import sys
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def test_imports():
    """测试依赖导入"""
    print("\n" + "="*60)
    print("📦 测试依赖导入...")
    print("="*60)
    
    try:
        import arxiv
        print("✅ arxiv")
    except ImportError:
        print("❌ arxiv - 请运行: pip install arxiv")
        return False
    
    try:
        import requests
        print("✅ requests")
    except ImportError:
        print("❌ requests - 请运行: pip install requests")
        return False
    
    try:
        from bs4 import BeautifulSoup
        print("✅ beautifulsoup4")
    except ImportError:
        print("❌ beautifulsoup4 - 请运行: pip install beautifulsoup4")
        return False
    
    try:
        import feedparser
        print("✅ feedparser")
    except ImportError:
        print("❌ feedparser - 请运行: pip install feedparser")
        return False
    
    print("\n✅ 所有依赖已安装\n")
    return True


def test_arxiv_fetcher():
    """测试 arXiv 模块"""
    print("\n" + "="*60)
    print("🔬 测试 arXiv 模块...")
    print("="*60)
    
    try:
        from arxiv_fetcher import ArxivFetcher
        
        # 创建获取器（只获取1篇论文用于测试）
        fetcher = ArxivFetcher(categories=['cs.AI'], max_results=1)
        print("✅ ArxivFetcher 初始化成功")
        
        # 搜索论文
        print("🔍 搜索最新论文...")
        papers = fetcher.search_papers("machine learning", max_results=2)
        
        if papers:
            print(f"✅ 成功获取 {len(papers)} 篇论文")
            print(f"\n示例论文:")
            print(f"   标题: {papers[0]['title'][:60]}...")
            print(f"   链接: {papers[0]['arxiv_url']}")
        else:
            print("⚠️  未获取到论文（可能是网络问题）")
        
        return True
        
    except Exception as e:
        print(f"❌ arXiv 模块测试失败: {str(e)}")
        return False


def test_huggingface_fetcher():
    """测试 HuggingFace 模块"""
    print("\n" + "="*60)
    print("🤗 测试 HuggingFace 模块...")
    print("="*60)
    
    try:
        from huggingface_fetcher import HuggingFaceFetcher
        
        # 创建获取器
        fetcher = HuggingFaceFetcher(max_results=2)
        print("✅ HuggingFaceFetcher 初始化成功")
        
        # 获取论文
        print("🔍 获取热门论文...")
        papers = fetcher.fetch_daily_papers()
        
        if papers:
            print(f"✅ 成功获取 {len(papers)} 篇论文")
            print(f"\n示例论文:")
            print(f"   标题: {papers[0]['title'][:60]}...")
            print(f"   链接: {papers[0]['url']}")
        else:
            print("⚠️  未获取到论文（可能是网络问题或页面结构变化）")
        
        return True
        
    except Exception as e:
        print(f"❌ HuggingFace 模块测试失败: {str(e)}")
        return False


def test_main_module():
    """测试主模块"""
    print("\n" + "="*60)
    print("🚀 测试主模块...")
    print("="*60)
    
    try:
        from main import PaperDigest
        
        # 创建实例
        digest = PaperDigest()
        print("✅ PaperDigest 初始化成功")
        
        # 注意：这里不实际运行 fetch_all_papers，因为可能需要较长时间
        print("✅ 主模块加载正常")
        
        return True
        
    except Exception as e:
        print(f"❌ 主模块测试失败: {str(e)}")
        return False


def main():
    """运行所有测试"""
    print("\n" + "="*60)
    print("🧪 开始运行测试套件")
    print("="*60)
    
    results = []
    
    # 测试依赖
    results.append(("依赖导入", test_imports()))
    
    # 测试模块
    if results[0][1]:  # 只有依赖正常才继续测试
        results.append(("arXiv 模块", test_arxiv_fetcher()))
        results.append(("HuggingFace 模块", test_huggingface_fetcher()))
        results.append(("主模块", test_main_module()))
    
    # 输出结果
    print("\n" + "="*60)
    print("📊 测试结果汇总")
    print("="*60)
    
    for name, result in results:
        status = "✅ 通过" if result else "❌ 失败"
        print(f"{name}: {status}")
    
    all_passed = all(result for _, result in results)
    
    print("\n" + "="*60)
    if all_passed:
        print("🎉 所有测试通过！")
        print("="*60)
        print("\n下一步:")
        print("1. 运行 python main.py 获取今日论文")
        print("2. 编辑 config/sources.json 自定义配置")
        print("3. 部署到 OpenClaw")
    else:
        print("⚠️  部分测试失败，请检查错误信息")
        print("="*60)
        print("\n故障排除:")
        print("1. 确保已安装所有依赖: pip install -r requirements.txt")
        print("2. 检查网络连接")
        print("3. 查看详细错误信息")
    
    print("")
    
    return 0 if all_passed else 1


if __name__ == "__main__":
    sys.exit(main())