Ai News Digest

# Example Asset File

This placeholder represents where asset files would be stored.
Replace with actual asset files (templates, images, fonts, etc.) or delete if not needed.

Asset files are NOT intended to be loaded into context, but rather used within
the output Claude produces.

Example asset files from other skills:
- Brand guidelines: logo.png, slides_template.pptx
- Frontend builder: hello-world/ directory with HTML/React boilerplate
- Typography: custom-font.ttf, font-family.woff2
- Data: sample_data.csv, test_dataset.json

## Common Asset Types

- Templates: .pptx, .docx, boilerplate directories
- Images: .png, .jpg, .svg, .gif
- Fonts: .ttf, .otf, .woff, .woff2
- Boilerplate code: Project directories, starter files
- Icons: .ico, .svg
- Data files: .csv, .json, .xml, .yaml

Note: This is a text placeholder. Actual assets can be any file type.

{
  "obsidian": {
    "vault_path": "",
    "notes_dir": "AI News",
    "filename_format": "AI-News-{date}.md"
  }
}

{
  "sources": [
    {
      "name": "Hacker News AI",
      "type": "rss",
      "url": "https://hn.algolia.com/api/v1/search?query=AI&tags=story",
      "language": "en",
      "category": "hot"
    },
    {
      "name": "TechCrunch AI",
      "type": "rss",
      "url": "https://techcrunch.com/category/artificial-intelligence/feed/",
      "language": "en",
      "category": "industry"
    },
    {
      "name": "The Verge AI",
      "type": "rss",
      "url": "https://www.theverge.com/ai-artificial-intelligence/rss/index.xml",
      "language": "en",
      "category": "industry"
    },
    {
      "name": "OpenAI Blog",
      "type": "rss",
      "url": "https://openai.com/blog/rss.xml",
      "language": "en",
      "category": "hot"
    },
    {
      "name": "机器之心",
      "type": "rss",
      "url": "https://www.jiqizhixin.com/rss",
      "language": "zh",
      "category": "cn"
    },
    {
      "name": "量子位",
      "type": "rss",
      "url": "https://www.qbitai.com/feed",
      "language": "zh",
      "category": "cn"
    },
    {
      "name": "36氪快讯",
      "type": "rss",
      "url": "https://rsshub.app/36kr/newsflashes",
      "language": "zh",
      "category": "cn"
    },
    {
      "name": "新智元",
      "type": "rss",
      "url": "https://rsshub.app/aibase/news",
      "language": "zh",
      "category": "cn"
    }
  ],
  "settings": {
    "max_news_per_source": 20,
    "final_output_count": 15,
    "similarity_threshold": 0.7,
    "timeout": 10
  }
}

# AI News Digest Skill - Python Dependencies

# RSS Feed 解析
feedparser>=6.0.10

# HTML 解析和网页抓取
beautifulsoup4>=4.12.0
requests>=2.31.0

# 数据处理
numpy>=1.24.0

# 日期时间处理（标准库，但某些环境需要）
python-dateutil>=2.8.2

#!/usr/bin/env python3
"""
AI调用助手：通过创建临时文件让AI助手处理翻译和摘要任务
"""

import json
import os
import time
from typing import Dict, List

class AIHelper:
    """通过文件系统与AI助手交互"""
    
    def __init__(self):
        self.temp_dir = "/tmp/ai_helper"
        os.makedirs(self.temp_dir, exist_ok=True)
    
    def translate_batch(self, news_list: List[Dict]) -> List[Dict]:
        """批量翻译英文新闻"""
        request_file = f"{self.temp_dir}/translate_request.json"
        response_file = f"{self.temp_dir}/translate_response.json"
        
        # 准备请求数据
        request_data = {
            "task": "translate",
            "news": news_list
        }
        
        # 写入请求
        with open(request_file, 'w', encoding='utf-8') as f:
            json.dump(request_data, f, ensure_ascii=False, indent=2)
        
        print(f"\n{'='*60}")
        print("🤖 需要AI助手帮助！")
        print(f"{'='*60}")
        print(f"请求文件: {request_file}")
        print(f"响应文件: {response_file}")
        print(f"\n需要翻译 {len(news_list)} 条英文新闻")
        print("\n请AI助手读取请求文件，翻译新闻，并将结果写入响应文件。")
        print("响应格式：JSON数组，每个元素包含 title 和 summary")
        print(f"{'='*60}\n")
        
        # 等待响应文件（简化处理：假设AI会立即处理）
        # 在实际场景中，这里会是一个交互式流程
        
        return news_list  # 临时返回原数据
    
    def summarize_batch(self, news_list: List[Dict]) -> List[Dict]:
        """批量生成中文摘要"""
        request_file = f"{self.temp_dir}/summarize_request.json"
        response_file = f"{self.temp_dir}/summarize_response.json"
        
        # 过滤需要摘要的新闻
        need_summary = [news for news in news_list 
                        if not news.get('summary') or len(news.get('summary', '')) < 20]
        
        if not need_summary:
            return news_list
        
        # 准备请求数据
        request_data = {
            "task": "summarize",
            "news": need_summary
        }
        
        # 写入请求
        with open(request_file, 'w', encoding='utf-8') as f:
            json.dump(request_data, f, ensure_ascii=False, indent=2)
        
        print(f"\n{'='*60}")
        print("🤖 需要AI助手帮助！")
        print(f"{'='*60}")
        print(f"请求文件: {request_file}")
        print(f"响应文件: {response_file}")
        print(f"\n需要为 {len(need_summary)} 条中文新闻生成摘要")
        print("\n请AI助手读取请求文件，生成摘要，并将结果写入响应文件。")
        print("响应格式：JSON数组，每个元素包含 title 和 summary")
        print(f"{'='*60}\n")
        
        return news_list  # 临时返回原数据


# 直接使用内联AI调用的简化版本
def translate_news_inline(news: Dict) -> Dict:
    """
    直接返回翻译后的新闻（手动翻译版本）
    实际使用时，这里应该调用真实的AI API
    """
    # 这里可以集成任何AI服务：OpenAI, Claude, 本地模型等
    # 为了演示，我们提供一个框架
    
    title = news['title']
    summary = news.get('summary', '')
    
    # TODO: 调用实际AI服务
    # 示例：使用OpenAI API
    # response = openai.ChatCompletion.create(...)
    
    # 临时方案：返回原文（在实际使用时会被替换）
    return news


def generate_summary_inline(news: Dict) -> Dict:
    """
    直接生成摘要（手动版本）
    实际使用时，这里应该调用真实的AI API
    """
    if news.get('summary') and len(news['summary']) > 20:
        return news
    
    # TODO: 调用实际AI服务生成摘要
    
    # 临时方案：使用标题作为摘要
    if not news.get('summary'):
        news['summary'] = news['title'][:50] + "..."
    
    return news


if __name__ == '__main__':
    # 测试
    helper = AIHelper()
    
    test_news = [
        {
            'title': 'OpenAI releases GPT-5',
            'summary': 'OpenAI has announced GPT-5',
            'source': 'TechCrunch',
            'url': 'https://example.com'
        }
    ]
    
    result = helper.translate_batch(test_news)
    print(json.dumps(result, ensure_ascii=False, indent=2))

#!/usr/bin/env python3
"""
内容增强模块：翻译英文新闻，为中文新闻生成摘要
"""

import json
import subprocess
import os
import hashlib
import time
from typing import List, Dict

class ContentEnhancer:
    """使用AI增强新闻内容"""
    
    def __init__(self):
        pass
    
    def enhance_news(self, news_list: List[Dict]) -> List[Dict]:
        """
        批量增强新闻内容
        - 英文新闻：翻译标题和摘要为中文
        - 中文新闻：生成摘要
        """
        enhanced = []
        
        for news in news_list:
            try:
                # 判断是否为英文新闻（简单判断：标题是否包含中文）
                is_english = not self._contains_chinese(news['title'])
                
                if is_english:
                    # 翻译英文新闻
                    enhanced_news = self._translate_english_news(news)
                else:
                    # 为中文新闻生成摘要
                    enhanced_news = self._generate_chinese_summary(news)
                
                enhanced.append(enhanced_news)
                
            except Exception as e:
                print(f"  警告：处理新闻失败 - {news.get('title', 'Unknown')[:50]}... 错误：{e}")
                # 失败时保留原内容
                enhanced.append(news)
        
        return enhanced
    
    def _contains_chinese(self, text: str) -> bool:
        """检查文本是否包含中文字符"""
        for char in text:
            if '\u4e00' <= char <= '\u9fff':
                return True
        return False
    
    def _translate_english_news(self, news: Dict) -> Dict:
        """翻译英文新闻为中文"""
        prompt = f"""请将以下英文AI新闻翻译成中文，保持专业性和准确性：

标题：{news['title']}
摘要：{news.get('summary', '无摘要')}

请以JSON格式返回，格式如下：
{{
  "title": "翻译后的标题",
  "summary": "翻译后的摘要"
}}

只返回JSON，不要其他内容。"""
        
        result = self._call_ai(prompt)
        
        try:
            translated = json.loads(result)
            news['title'] = translated['title']
            news['summary'] = translated['summary']
        except:
            print(f"  警告：翻译失败，保留原文 - {news['title'][:50]}...")
        
        return news
    
    def _generate_chinese_summary(self, news: Dict) -> Dict:
        """为中文新闻生成摘要"""
        # 如果已有摘要且长度合适，跳过
        if news.get('summary') and len(news['summary']) > 20:
            return news
        
        prompt = f"""请为以下中文AI新闻生成一个简洁的摘要（30-50字）：

标题：{news['title']}
原摘要：{news.get('summary', '无')}

请直接返回摘要文本，不要其他内容。"""
        
        result = self._call_ai(prompt)
        
        # 清理结果（去掉可能的引号等）
        summary = result.strip().strip('"').strip("'")
        
        if len(summary) > 10:  # 基本验证
            news['summary'] = summary
        
        return news
    
    def _call_ai(self, prompt: str) -> str:
        """
        调用AI - 通过创建请求文件，让AI助手处理
        """
        # 生成唯一ID
        request_id = hashlib.md5(f"{time.time()}".encode()).hexdigest()[:8]
        request_file = f"/tmp/ai_request_{request_id}.txt"
        response_file = f"/tmp/ai_response_{request_id}.json"
        
        # 写入请求
        with open(request_file, 'w', encoding='utf-8') as f:
            f.write(prompt)
        
        print(f"\n  💬 AI请求已创建: {request_file}")
        print(f"  📝 等待响应文件: {response_file}")
        
        # 等待响应文件（最多30秒）
        max_wait = 30
        for i in range(max_wait):
            if os.path.exists(response_file):
                with open(response_file, 'r', encoding='utf-8') as f:
                    result = f.read()
                
                # 清理临时文件
                try:
                    os.remove(request_file)
                    os.remove(response_file)
                except:
                    pass
                
                return result
            
            time.sleep(1)
        
        # 超时返回默认值
        print(f"  ⚠️  AI响应超时")
        return '{"title": "翻译超时", "summary": "请稍后重试"}'
    
    def enhance_with_batch_ai(self, news_list: List[Dict]) -> List[Dict]:
        """
        使用批量AI调用增强内容（更高效）
        将所有需要处理的新闻一次性发送给AI
        """
        # 分类新闻
        english_news = []
        chinese_news = []
        
        for i, news in enumerate(news_list):
            is_english = not self._contains_chinese(news['title'])
            if is_english:
                english_news.append((i, news))
            else:
                chinese_news.append((i, news))
        
        # 批量翻译英文新闻
        if english_news:
            print(f"  翻译 {len(english_news)} 条英文新闻...")
            translated = self._batch_translate(english_news)
            for idx, enhanced in translated:
                news_list[idx] = enhanced
        
        # 批量生成中文摘要
        if chinese_news:
            print(f"  生成 {len(chinese_news)} 条中文摘要...")
            summarized = self._batch_summarize(chinese_news)
            for idx, enhanced in summarized:
                news_list[idx] = enhanced
        
        return news_list
    
    def _batch_translate(self, news_items: List[tuple]) -> List[tuple]:
        """批量翻译英文新闻"""
        # 构建批量prompt
        items_text = []
        for idx, news in news_items:
            items_text.append(f"""
新闻 {idx}:
标题: {news['title']}
摘要: {news.get('summary', '无')}
""")
        
        prompt = f"""请将以下{len(news_items)}条英文AI新闻翻译成中文，保持专业性和准确性。

{''.join(items_text)}

请以JSON数组格式返回，每个元素包含index、title、summary：
[
  {{"index": 0, "title": "翻译后的标题", "summary": "翻译后的摘要"}},
  ...
]

只返回JSON数组，不要其他内容。"""
        
        # TODO: 调用实际AI
        result = self._call_ai_batch(prompt)
        
        # 解析结果并更新新闻
        enhanced = []
        try:
            translations = json.loads(result)
            for trans in translations:
                idx = trans['index']
                original_idx, news = news_items[idx]
                news['title'] = trans['title']
                news['summary'] = trans['summary']
                enhanced.append((original_idx, news))
        except:
            # 失败时返回原文
            enhanced = news_items
        
        return enhanced
    
    def _batch_summarize(self, news_items: List[tuple]) -> List[tuple]:
        """批量生成中文摘要"""
        # 过滤掉已有合适摘要的新闻
        need_summary = [(idx, news) for idx, news in news_items 
                        if not news.get('summary') or len(news.get('summary', '')) < 20]
        
        if not need_summary:
            return news_items
        
        # 构建批量prompt
        items_text = []
        for idx, news in need_summary:
            items_text.append(f"""
新闻 {idx}:
标题: {news['title']}
原摘要: {news.get('summary', '无')}
""")
        
        prompt = f"""请为以下{len(need_summary)}条中文AI新闻生成简洁的摘要（每条30-50字）。

{''.join(items_text)}

请以JSON数组格式返回，每个元素包含index和summary：
[
  {{"index": 0, "summary": "摘要内容"}},
  ...
]

只返回JSON数组，不要其他内容。"""
        
        # TODO: 调用实际AI
        result = self._call_ai_batch(prompt)
        
        # 解析结果并更新新闻
        enhanced = []
        try:
            summaries = json.loads(result)
            summary_dict = {s['index']: s['summary'] for s in summaries}
            
            for idx, news in news_items:
                if idx in summary_dict:
                    news['summary'] = summary_dict[idx]
                enhanced.append((idx, news))
        except:
            # 失败时返回原文
            enhanced = news_items
        
        return enhanced
    
    def _call_ai_batch(self, prompt: str) -> str:
        """批量调用AI"""
        # TODO: 实现实际的AI调用
        return '[]'


if __name__ == '__main__':
    # 测试代码
    enhancer = ContentEnhancer()
    
    test_news = [
        {
            'title': 'OpenAI releases GPT-5',
            'summary': 'OpenAI has announced the release of GPT-5, their latest language model.',
            'source': 'TechCrunch',
            'url': 'https://example.com'
        },
        {
            'title': '字节跳动发布新AI模型',
            'summary': '',
            'source': '机器之心',
            'url': 'https://example.com'
        }
    ]
    
    enhanced = enhancer.enhance_news(test_news)
    print(json.dumps(enhanced, ensure_ascii=False, indent=2))

#!/usr/bin/env python3
"""
Example helper script for ai-news-digest

This is a placeholder script that can be executed directly.
Replace with actual implementation or delete if not needed.

Example real scripts from other skills:
- pdf/scripts/fill_fillable_fields.py - Fills PDF form fields
- pdf/scripts/convert_pdf_to_images.py - Converts PDF pages to images
"""

def main():
    print("This is an example script for ai-news-digest")
    # TODO: Add actual script logic here
    # This could be data processing, file conversion, API calls, etc.

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
新闻抓取模块
支持RSS和网页抓取两种方式
"""

import os
import requests
import feedparser
import json
import time
from datetime import datetime, timedelta
from bs4 import BeautifulSoup
from typing import List, Dict
from pathlib import Path
import sys

class NewsFetcher:
    def __init__(self, config_path: str):
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
        self.timeout = self.config['settings']['timeout']
        self.max_per_source = self.config['settings']['max_news_per_source']
        
    def fetch_all(self, hours: int = 24) -> List[Dict]:
        """抓取所有新闻源"""
        all_news = []
        cutoff_time = datetime.now() - timedelta(hours=hours)
        
        for source in self.config['sources']:
            try:
                print(f"正在抓取: {source['name']}...", end=' ')
                news = self._fetch_rss(source, cutoff_time)
                all_news.extend(news)
                print(f"✓ {len(news)} 条")
                time.sleep(1)  # 避免请求过快
            except Exception as e:
                print(f"✗ {str(e)}")
                continue
                
        return all_news
    
    def _fetch_rss(self, source: Dict, cutoff_time: datetime) -> List[Dict]:
        """抓取RSS源"""
        news_list = []
        
        # 特殊处理Hacker News API
        if 'hn.algolia.com' in source['url']:
            response = requests.get(source['url'], timeout=self.timeout)
            data = response.json()
            
            for hit in data.get('hits', [])[:self.max_per_source]:
                created_at = datetime.fromtimestamp(hit.get('created_at_i', 0))
                if created_at < cutoff_time:
                    continue
                    
                news_list.append({
                    'title': hit.get('title', ''),
                    'url': hit.get('url', f"https://news.ycombinator.com/item?id={hit.get('objectID')}"),
                    'summary': hit.get('story_text', '')[:200] if hit.get('story_text') else '',
                    'published': created_at,
                    'source': source['name'],
                    'language': source['language'],
                    'category': source['category'],
                    'score': hit.get('points', 0)
                })
        else:
            # 标准RSS处理
            feed = feedparser.parse(source['url'])
            
            for entry in feed.entries[:self.max_per_source]:
                # 解析时间
                pub_time = None
                if hasattr(entry, 'published_parsed') and entry.published_parsed:
                    pub_time = datetime(*entry.published_parsed[:6])
                elif hasattr(entry, 'updated_parsed') and entry.updated_parsed:
                    pub_time = datetime(*entry.updated_parsed[:6])
                else:
                    pub_time = datetime.now()
                
                if pub_time < cutoff_time:
                    continue
                
                # 提取摘要
                summary = ''
                if hasattr(entry, 'summary'):
                    soup = BeautifulSoup(entry.summary, 'html.parser')
                    summary = soup.get_text()[:200]
                
                news_list.append({
                    'title': entry.get('title', ''),
                    'url': entry.get('link', ''),
                    'summary': summary,
                    'published': pub_time,
                    'source': source['name'],
                    'language': source['language'],
                    'category': source['category'],
                    'score': 0
                })
        
        return news_list

if __name__ == '__main__':
    import argparse
    
    # 获取脚本所在目录的绝对路径
    SCRIPT_DIR = Path(__file__).parent.absolute()
    DEFAULT_CONFIG = SCRIPT_DIR.parent / 'config' / 'sources.json'
    
    parser = argparse.ArgumentParser(description='抓取AI新闻')
    parser.add_argument('--hours', type=int, default=24, help='时间范围（小时）')
    parser.add_argument('--config', type=str, default=str(DEFAULT_CONFIG), help='配置文件路径')
    args = parser.parse_args()
    
    fetcher = NewsFetcher(args.config)
    news = fetcher.fetch_all(hours=args.hours)
    
    print(f"\n总共抓取 {len(news)} 条新闻")
    
    # 输出到JSON文件
    output_file = f'/tmp/raw_news_{int(time.time())}.json'
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(news, f, ensure_ascii=False, indent=2, default=str)
    print(f"已保存到: {output_file}")

#!/usr/bin/env python3
"""
Digest生成和发送模块
"""

import json
import os
import subprocess
import shutil
from datetime import datetime
from typing import Dict
from pathlib import Path
import argparse

class DigestGenerator:
    def __init__(self, output_config_path: str):
        with open(output_config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
    
    def generate(self, news_data: Dict, date_str: str = None) -> str:
        """生成Markdown格式的digest"""
        if not date_str:
            date_str = datetime.now().strftime('%Y-%m-%d')
        
        # 生成Markdown
        md_content = self._generate_markdown(news_data, date_str)
        
        return md_content
    
    def _generate_markdown(self, news_data: Dict, date_str: str) -> str:
        """生成Markdown格式"""
        lines = [f"# AI新闻 Digest ({date_str})\n"]
        
        # 热点新闻
        if news_data.get('hot'):
            lines.append("## 🔥 热点新闻\n")
            for i, news in enumerate(news_data['hot'], 1):
                lines.append(f"{i}. **{news['title']}**")
                lines.append(f"   - 来源：{news['source']}")
                lines.append(f"   - 发布时间：{self._format_time(news['published'])}")
                if news.get('summary'):
                    lines.append(f"   - 摘要：{news['summary']}")
                lines.append(f"   - [阅读原文]({news['url']})\n")
        
        # 行业动态
        if news_data.get('industry'):
            lines.append("## 📊 行业动态\n")
            for i, news in enumerate(news_data['industry'], 1):
                lines.append(f"{i}. **{news['title']}**")
                lines.append(f"   - 来源：{news['source']}")
                lines.append(f"   - 发布时间：{self._format_time(news['published'])}")
                if news.get('summary'):
                    lines.append(f"   - 摘要：{news['summary']}")
                lines.append(f"   - [阅读原文]({news['url']})\n")
        
        # 中文资讯
        if news_data.get('cn'):
            lines.append("## 🌏 中文资讯\n")
            for i, news in enumerate(news_data['cn'], 1):
                lines.append(f"{i}. **{news['title']}**")
                lines.append(f"   - 来源：{news['source']}")
                lines.append(f"   - 发布时间：{self._format_time(news['published'])}")
                if news.get('summary'):
                    lines.append(f"   - 摘要：{news['summary']}")
                lines.append(f"   - [阅读原文]({news['url']})\n")
        
        lines.append(f"\n---\n*由AI News Digest自动生成于 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}*")
        
        return '\n'.join(lines)
    
    def _format_time(self, time_obj) -> str:
        """格式化时间"""
        if isinstance(time_obj, str):
            try:
                time_obj = datetime.fromisoformat(time_obj)
            except:
                return time_obj
        
        now = datetime.now()
        diff = now - time_obj
        
        if diff.days > 0:
            return f"{diff.days}天前"
        elif diff.seconds > 3600:
            return f"{diff.seconds // 3600}小时前"
        else:
            return f"{diff.seconds // 60}分钟前"
    
    def save(self, md_content: str, date_str: str):
        """保存到文件和Obsidian"""
        # 1. 保存Markdown文件
        md_filename = f"AI-News-{date_str}.md"
        md_path = f"/tmp/{md_filename}"
        with open(md_path, 'w', encoding='utf-8') as f:
            f.write(md_content)
        print(f"✓ Markdown已保存: {md_path}")
        
        # 2. 保存到Obsidian（如果配置了）
        if 'obsidian' in self.config:
            obsidian_config = self.config['obsidian']
            vault_path = obsidian_config.get('vault_path')
            notes_dir = obsidian_config.get('notes_dir', 'AI News')
            
            if vault_path:
                full_notes_dir = os.path.join(vault_path, notes_dir)
                os.makedirs(full_notes_dir, exist_ok=True)
                
                obsidian_file = os.path.join(full_notes_dir, md_filename)
                shutil.copy(md_path, obsidian_file)
                print(f"✓ 已存入Obsidian: {obsidian_file}")
        
        return md_path

def main():
    # 获取脚本所在目录的绝对路径
    SCRIPT_DIR = Path(__file__).parent.absolute()
    CONFIG_DIR = SCRIPT_DIR.parent / 'config'
    
    parser = argparse.ArgumentParser(description='生成AI新闻Digest')
    parser.add_argument('--hours', type=int, help='时间范围（小时）')
    parser.add_argument('--days', type=int, help='时间范围（天）')
    args = parser.parse_args()
    
    # 确定时间范围
    hours = args.hours if args.hours else (args.days * 24 if args.days else 24)
    
    print(f"=== AI News Digest Generator ===")
    print(f"时间范围: 过去 {hours} 小时\n")
    
    # 1. 抓取新闻
    print("[1/4] 抓取新闻...")
    from fetch_news import NewsFetcher
    fetcher = NewsFetcher(str(CONFIG_DIR / 'sources.json'))
    raw_news = fetcher.fetch_all(hours=hours)
    print(f"  抓取完成: {len(raw_news)} 条\n")
    
    # 2. 处理新闻
    print("[2/4] 处理新闻（去重、排序）...")
    from process_news import NewsProcessor
    processor = NewsProcessor()
    processed_news = processor.process(raw_news, max_output=15)
    print(f"  处理完成\n")
    
    # 2.5. 增强内容（翻译英文、生成中文摘要）
    print("[2.5/4] 增强内容（翻译+摘要）...")
    from enhance_content import ContentEnhancer
    enhancer = ContentEnhancer()
    
    # 对每个分类的新闻进行增强
    for category in ['hot', 'industry', 'cn']:
        if category in processed_news and processed_news[category]:
            print(f"  处理 {category} 分类...")
            processed_news[category] = enhancer.enhance_news(processed_news[category])
    print(f"  内容增强完成\n")
    
    # 3. 生成digest
    print("[3/4] 生成Markdown...")
    generator = DigestGenerator(str(CONFIG_DIR / 'output.json'))
    date_str = datetime.now().strftime('%Y-%m-%d')
    md_content = generator.generate(processed_news, date_str)
    print(f"  生成完成\n")
    
    # 4. 保存
    print("[4/4] 保存文件...")
    output_path = generator.save(md_content, date_str)
    print(f"\n=== 完成 ===")
    print(f"输出文件: {output_path}")

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
新闻处理模块
负责去重、排序、汇总
"""

import json
from typing import List, Dict
from difflib import SequenceMatcher
from datetime import datetime

class NewsProcessor:
    def __init__(self, similarity_threshold: float = 0.7):
        self.similarity_threshold = similarity_threshold
    
    def process(self, news_list: List[Dict], max_output: int = 15) -> Dict:
        """处理新闻：去重 -> 排序 -> 分类"""
        # 1. 去重
        unique_news = self._deduplicate(news_list)
        print(f"去重后: {len(unique_news)} 条")
        
        # 2. 排序（综合评分）
        scored_news = self._calculate_scores(unique_news)
        sorted_news = sorted(scored_news, key=lambda x: x['final_score'], reverse=True)
        
        # 3. 分类输出
        result = {
            'hot': [],      # 热点新闻（英文，高分）
            'industry': [], # 行业动态（英文，中等分）
            'cn': []        # 中文资讯
        }
        
        for news in sorted_news[:max_output]:
            category = news['category']
            if category in result:
                result[category].append(news)
            else:
                # 默认归类
                if news['language'] == 'zh':
                    result['cn'].append(news)
                else:
                    result['industry'].append(news)
        
        return result
    
    def _deduplicate(self, news_list: List[Dict]) -> List[Dict]:
        """基于标题相似度去重"""
        unique = []
        
        for news in news_list:
            is_duplicate = False
            for existing in unique:
                similarity = self._similarity(news['title'], existing['title'])
                if similarity > self.similarity_threshold:
                    # 保留分数更高的那个
                    if news.get('score', 0) > existing.get('score', 0):
                        unique.remove(existing)
                        unique.append(news)
                    is_duplicate = True
                    break
            
            if not is_duplicate:
                unique.append(news)
        
        return unique
    
    def _similarity(self, s1: str, s2: str) -> float:
        """计算两个字符串的相似度"""
        return SequenceMatcher(None, s1.lower(), s2.lower()).ratio()
    
    def _calculate_scores(self, news_list: List[Dict]) -> List[Dict]:
        """计算综合评分"""
        now = datetime.now()
        
        for news in news_list:
            # 基础分数
            score = 0
            
            # 1. 热度分（原始分数，如HN的points）
            score += news.get('score', 0) * 0.3
            
            # 2. 新鲜度分（越新越高）
            pub_time = news.get('published', now)
            if isinstance(pub_time, str):
                try:
                    pub_time = datetime.fromisoformat(pub_time)
                except:
                    pub_time = now
            
            hours_ago = (now - pub_time).total_seconds() / 3600
            freshness = max(0, 100 - hours_ago)  # 24小时内从100降到0
            score += freshness * 0.4
            
            # 3. 来源权威性分
            source_weights = {
                'OpenAI Blog': 50,
                'Hacker News AI': 40,
                'TechCrunch AI': 35,
                'The Verge AI': 30,
                '机器之心': 35,
                '量子位': 30,
                'AI科技评论': 30,
                '36氪AI': 25
            }
            score += source_weights.get(news['source'], 20) * 0.3
            
            news['final_score'] = score
        
        return news_list

if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser(description='处理AI新闻')
    parser.add_argument('--input', type=str, required=True, help='输入JSON文件')
    parser.add_argument('--output', type=str, required=True, help='输出JSON文件')
    parser.add_argument('--max', type=int, default=15, help='最大输出数量')
    args = parser.parse_args()
    
    # 读取原始新闻
    with open(args.input, 'r', encoding='utf-8') as f:
        raw_news = json.load(f)
    
    # 处理
    processor = NewsProcessor()
    processed = processor.process(raw_news, max_output=args.max)
    
    # 保存
    with open(args.output, 'w', encoding='utf-8') as f:
        json.dump(processed, f, ensure_ascii=False, indent=2, default=str)
    
    print(f"处理完成，已保存到: {args.output}")
    print(f"  热点新闻: {len(processed['hot'])} 条")
    print(f"  行业动态: {len(processed['industry'])} 条")
    print(f"  中文资讯: {len(processed['cn'])} 条")

#!/usr/bin/env python3
"""
Standalone test script for AI News Digest
Tests each component independently and shows results
"""

import sys
import feedparser
import requests
from datetime import datetime, timedelta
from difflib import SequenceMatcher

def test_rss_feed(url, name):
    """Test a single RSS feed"""
    print(f"\n{'='*60}")
    print(f"Testing: {name}")
    print(f"URL: {url}")
    print(f"{'='*60}")
    
    try:
        if 'hn.algolia.com' in url:
            # Hacker News API
            response = requests.get(url, timeout=10)
            data = response.json()
            articles = data.get('hits', [])
            print(f"✓ SUCCESS: Fetched {len(articles)} articles")
            
            # Show first 3
            for i, article in enumerate(articles[:3], 1):
                title = article.get('title', 'No title')
                points = article.get('points', 0)
                print(f"  {i}. {title} ({points} points)")
                
        else:
            # RSS feed
            feed = feedparser.parse(url)
            entries = feed.entries
            print(f"✓ SUCCESS: Fetched {len(entries)} articles")
            
            # Show first 3
            for i, entry in enumerate(entries[:3], 1):
                title = entry.get('title', 'No title')
                print(f"  {i}. {title}")
                
        return True
        
    except Exception as e:
        print(f"✗ FAILED: {str(e)}")
        return False

def test_deduplication():
    """Test similarity-based deduplication"""
    print(f"\n{'='*60}")
    print("Testing: Deduplication Algorithm")
    print(f"{'='*60}")
    
    test_titles = [
        "OpenAI Releases GPT-5 with Advanced Reasoning",
        "OpenAI Launches GPT-5 With Enhanced Reasoning Capabilities",  # Similar
        "Google Announces Gemini 2.0",
        "Microsoft Invests $10B in AI Startup",
        "OpenAI's GPT-5: A New Era of AI Reasoning",  # Similar to first
    ]
    
    def similarity(a, b):
        return SequenceMatcher(None, a.lower(), b.lower()).ratio()
    
    threshold = 0.70
    unique_titles = []
    duplicates = []
    
    for title in test_titles:
        is_duplicate = False
        for existing in unique_titles:
            sim_score = similarity(title, existing)
            if sim_score > threshold:
                is_duplicate = True
                duplicates.append((title, existing, sim_score))
                break
        
        if not is_duplicate:
            unique_titles.append(title)
    
    print(f"\nOriginal articles: {len(test_titles)}")
    print(f"After deduplication: {len(unique_titles)}")
    print(f"Duplicates removed: {len(duplicates)}")
    
    print("\n✓ Unique articles:")
    for i, title in enumerate(unique_titles, 1):
        print(f"  {i}. {title}")
    
    if duplicates:
        print("\n✗ Duplicates detected:")
        for dup, original, score in duplicates:
            print(f"  - '{dup}'")
            print(f"    → Similar to: '{original}'")
            print(f"    → Similarity: {score:.2%}")

def test_scoring():
    """Test article scoring algorithm"""
    print(f"\n{'='*60}")
    print("Testing: Scoring Algorithm")
    print(f"{'='*60}")
    
    test_articles = [
        {
            'title': 'OpenAI Announces GPT-5 Breakthrough',
            'source': 'OpenAI Blog',
            'published': datetime.now() - timedelta(hours=2),
        },
        {
            'title': 'Minor Update to TensorFlow Documentation',
            'source': 'GitHub',
            'published': datetime.now() - timedelta(days=3),
        },
        {
            'title': 'AGI研究新突破：清华团队发表Nature论文',
            'source': '机器之心',
            'published': datetime.now() - timedelta(hours=5),
        },
    ]
    
    def calculate_score(article):
        score = 0.0
        
        # Recency (0-3 points)
        hours_ago = (datetime.now() - article['published']).total_seconds() / 3600
        if hours_ago < 6:
            score += 3.0
        elif hours_ago < 24:
            score += 2.0
        else:
            score += 1.0
        
        # Hot keywords (0-4 points)
        hot_keywords = ['GPT', 'OpenAI', 'breakthrough', 'AGI', '突破', 'Nature']
        title_lower = article['title'].lower()
        for keyword in hot_keywords:
            if keyword.lower() in title_lower:
                score += 1.0
        
        # Source authority (0-3 points)
        authority_map = {
            'OpenAI Blog': 3.0,
            'Hacker News': 2.5,
            '机器之心': 2.5,
            'TechCrunch': 2.0,
        }
        score += authority_map.get(article['source'], 1.0)
        
        return min(score, 10.0)
    
    print("\nArticle Scores:")
    scored = [(calculate_score(a), a) for a in test_articles]
    scored.sort(reverse=True, key=lambda x: x[0])
    
    for score, article in scored:
        hours_ago = (datetime.now() - article['published']).total_seconds() / 3600
        print(f"\n  Score: {score:.1f}/10")
        print(f"  Title: {article['title']}")
        print(f"  Source: {article['source']}")
        print(f"  Published: {hours_ago:.1f} hours ago")

def main():
    """Run all tests"""
    print("=" * 60)
    print("AI NEWS DIGEST - COMPONENT TESTS")
    print("=" * 60)
    
    # Test news sources
    sources = [
        ("https://hn.algolia.com/api/v1/search_by_date?tags=story&query=AI", "Hacker News API"),
        ("https://techcrunch.com/category/artificial-intelligence/feed/", "TechCrunch AI"),
        ("https://www.jiqizhixin.com/rss", "机器之心"),
        ("https://www.qbitai.com/feed", "量子位"),
    ]
    
    results = {}
    for url, name in sources:
        results[name] = test_rss_feed(url, name)
    
    # Test algorithms
    test_deduplication()
    test_scoring()
    
    # Summary
    print(f"\n{'='*60}")
    print("TEST SUMMARY")
    print(f"{'='*60}")
    
    success_count = sum(1 for v in results.values() if v)
    total_count = len(results)
    
    print(f"\nNews Sources: {success_count}/{total_count} working")
    for name, success in results.items():
        status = "✓" if success else "✗"
        print(f"  {status} {name}")
    
    print("\n✓ Deduplication: Working")
    print("✓ Scoring: Working")
    
    if success_count == total_count:
        print("\n🎉 All tests passed!")
        return 0
    else:
        print(f"\n⚠️  {total_count - success_count} source(s) failed")
        return 1

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

#!/usr/bin/env python3
"""
测试版本：只生成Markdown，不发邮件
"""

import sys
import os
sys.path.insert(0, '/data/workspace/my_skills/ai-news-digest/scripts')

from datetime import datetime
from fetch_news import NewsFetcher
from process_news import NewsProcessor
from generate_digest import DigestGenerator

def test_digest(hours=48):
    print(f"=== AI News Digest 测试 ===")
    print(f"时间范围: 过去 {hours} 小时\n")
    
    # 1. 抓取
    print("[1/3] 抓取新闻...")
    fetcher = NewsFetcher('/data/workspace/my_skills/ai-news-digest/config/sources.json')
    raw_news = fetcher.fetch_all(hours=hours)
    print(f"  抓取完成: {len(raw_news)} 条\n")
    
    if len(raw_news) == 0:
        print("没有抓取到新闻，退出测试")
        return
    
    # 2. 处理
    print("[2/3] 处理新闻...")
    processor = NewsProcessor()
    processed_news = processor.process(raw_news, max_output=15)
    print(f"  热点: {len(processed_news['hot'])} 条")
    print(f"  行业: {len(processed_news['industry'])} 条")
    print(f"  中文: {len(processed_news['cn'])} 条\n")
    
    # 3. 生成
    print("[3/3] 生成Digest...")
    generator = DigestGenerator('/data/workspace/my_skills/ai-news-digest/config/output.json')
    date_str = datetime.now().strftime('%Y-%m-%d')
    md_content, html_content = generator.generate(processed_news, date_str)
    
    # 保存测试文件
    test_md = f"/tmp/test-digest-{date_str}.md"
    with open(test_md, 'w', encoding='utf-8') as f:
        f.write(md_content)
    print(f"  ✓ Markdown已保存: {test_md}")
    
    test_html = f"/tmp/test-digest-{date_str}.html"
    with open(test_html, 'w', encoding='utf-8') as f:
        f.write(html_content)
    print(f"  ✓ HTML已保存: {test_html}")
    
    print("\n=== 测试完成 ===")
    print(f"请查看: {test_md}")

if __name__ == '__main__':
    test_digest(hours=48)