app.py

# -*- coding: utf-8 -*-
"""
AI 뉴스 & 허깅페이스 트렌딩 분석 시스템
- AI Times 뉴스 크롤링 및 카테고리 분류
- 허깅페이스 모델/스페이스 트렌딩 정보 수집
- Fireworks AI (Qwen) 를 통한 뉴스 분석
- Brave Search를 통한 팩트 체크
"""

import requests
from bs4 import BeautifulSoup
import json
from datetime import datetime
from typing import List, Dict, Optional
import time
import re


class AINewsAnalyzer:
    def __init__(self, fireworks_api_key: str, brave_api_key: str):
        """
        Args:
            fireworks_api_key: Fireworks AI API 키
            brave_api_key: Brave Search API 키
        """
        self.fireworks_api_key = fireworks_api_key
        self.brave_api_key = brave_api_key
        
        # 뉴스 카테고리 정의
        self.categories = {
            "산업동향": ["산업", "기업", "투자", "인수", "파트너십", "시장"],
            "기술혁신": ["기술", "모델", "알고리즘", "개발", "연구", "논문"],
            "제품출시": ["출시", "공개", "발표", "서비스", "제품"],
            "정책규제": ["규제", "정책", "법", "정부", "제재"],
            "보안이슈": ["보안", "취약점", "해킹", "위험", "프라이버시"],
        }
        
        self.huggingface_data = {
            "models": [],
            "spaces": []
        }
        
        self.news_data = []
    
    def fetch_aitimes_news(self, urls: List[str]) -> List[Dict]:
        """AI Times 뉴스 크롤링"""
        all_news = []
        
        for url in urls:
            try:
                print(f"📰 뉴스 크롤링 중: {url}")
                response = requests.get(url, headers={
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
                })
                soup = BeautifulSoup(response.content, 'html.parser')
                
                # 뉴스 기사 추출 (실제 구조에 맞게 조정 필요)
                articles = []
                
                # 제목과 링크가 있는 a 태그 찾기
                for link in soup.find_all('a', href=True):
                    if '/news/articleView.html' in link['href']:
                        title = link.get_text(strip=True)
                        article_url = link['href']
                        
                        if not article_url.startswith('http'):
                            article_url = 'https://www.aitimes.com' + article_url
                        
                        # 날짜 추출 (형제 요소에서)
                        date_text = ""
                        parent = link.parent
                        if parent:
                            date_elem = parent.find(text=re.compile(r'\d{2}-\d{2}'))
                            if date_elem:
                                date_text = date_elem.strip()
                        
                        if title and len(title) > 10:
                            articles.append({
                                'title': title,
                                'url': article_url,
                                'date': date_text,
                                'source': 'AI Times'
                            })
                
                all_news.extend(articles[:10])  # 상위 10개만
                time.sleep(1)  # 크롤링 예의
                
            except Exception as e:
                print(f"❌ 크롤링 오류: {e}")
        
        return all_news
    
    def fetch_huggingface_trending(self) -> Dict:
        """허깅페이스 트렌딩 모델 및 스페이스 수집"""
        print("🤗 허깅페이스 트렌딩 정보 수집 중...")
        
        # 모델 트렌딩
        try:
            models_url = "https://huggingface.co/api/models"
            params = {
                'sort': 'trending',
                'limit': 30
            }
            
            response = requests.get(models_url, params=params, timeout=10)
            if response.status_code == 200:
                models = response.json()
                
                for model in models[:30]:
                    self.huggingface_data['models'].append({
                        'name': model.get('id', 'Unknown'),
                        'downloads': model.get('downloads', 0),
                        'likes': model.get('likes', 0),
                        'task': model.get('pipeline_tag', 'N/A'),
                        'url': f"https://huggingface.co/{model.get('id', '')}"
                    })
                
                print(f"✅ {len(self.huggingface_data['models'])}개 트렌딩 모델 수집 완료")
        
        except Exception as e:
            print(f"❌ 모델 수집 오류: {e}")
        
        # 스페이스 트렌딩 (웹 크롤링)
        try:
            spaces_url = "https://huggingface.co/spaces"
            response = requests.get(spaces_url, headers={
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }, timeout=10)
            
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # 스페이스 링크 추출
            space_count = 0
            for link in soup.find_all('a', href=True):
                if '/spaces/' in link['href'] and space_count < 30:
                    space_name = link['href'].replace('/spaces/', '')
                    if '/' in space_name and len(space_name) > 3:
                        title = link.get_text(strip=True)
                        if title:
                            self.huggingface_data['spaces'].append({
                                'name': space_name,
                                'title': title[:100],
                                'url': f"https://huggingface.co{link['href']}"
                            })
                            space_count += 1
            
            print(f"✅ {len(self.huggingface_data['spaces'])}개 트렌딩 스페이스 수집 완료")
        
        except Exception as e:
            print(f"❌ 스페이스 수집 오류: {e}")
        
        return self.huggingface_data
    
    def categorize_news(self, news_list: List[Dict]) -> List[Dict]:
        """뉴스 카테고리 분류"""
        for news in news_list:
            title = news['title'].lower()
            news['category'] = "기타"
            
            for category, keywords in self.categories.items():
                if any(keyword in title for keyword in keywords):
                    news['category'] = category
                    break
        
        return news_list
    
    def analyze_with_qwen(self, text: str, instruction: str) -> str:
        """Fireworks AI Qwen 모델을 사용한 분석"""
        url = "https://api.fireworks.ai/inference/v1/chat/completions"
        
        payload = {
            "model": "accounts/fireworks/models/qwen3-235b-a22b-instruct-2507",
            "max_tokens": 4096,
            "top_p": 1,
            "top_k": 40,
            "presence_penalty": 0,
            "frequency_penalty": 0,
            "temperature": 0.6,
            "messages": [
                {
                    "role": "system",
                    "content": "당신은 AI 뉴스를 초등학생도 이해할 수 있게 쉽게 설명하는 전문가입니다."
                },
                {
                    "role": "user",
                    "content": f"{instruction}\n\n뉴스: {text}"
                }
            ]
        }
        
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.fireworks_api_key}"
        }
        
        try:
            response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=30)
            
            if response.status_code == 200:
                result = response.json()
                return result['choices'][0]['message']['content']
            else:
                return f"분석 실패 (상태 코드: {response.status_code})"
        
        except Exception as e:
            return f"분석 오류: {str(e)}"
    
    def fact_check_with_brave(self, query: str) -> List[Dict]:
        """Brave Search를 통한 팩트 체크"""
        url = "https://api.search.brave.com/res/v1/web/search"
        
        headers = {
            "Accept": "application/json",
            "X-Subscription-Token": self.brave_api_key
        }
        
        params = {
            "q": query,
            "count": 5,
            "text_decorations": False,
            "search_lang": "ko"
        }
        
        try:
            response = requests.get(url, headers=headers, params=params, timeout=10)
            
            if response.status_code == 200:
                data = response.json()
                results = []
                
                if 'web' in data and 'results' in data['web']:
                    for item in data['web']['results'][:3]:
                        results.append({
                            'title': item.get('title', ''),
                            'description': item.get('description', ''),
                            'url': item.get('url', '')
                        })
                
                return results
            else:
                return []
        
        except Exception as e:
            print(f"❌ Brave Search 오류: {e}")
            return []
    
    def generate_report(self, news_list: List[Dict], analyze_news: bool = True) -> str:
        """종합 리포트 생성"""
        report = []
        report.append("=" * 80)
        report.append("📊 AI 뉴스 & 허깅페이스 트렌딩 종합 리포트")
        report.append(f"📅 생성일시: {datetime.now().strftime('%Y년 %m월 %d일 %H:%M')}")
        report.append("=" * 80)
        report.append("")
        
        # 1. 카테고리별 뉴스 분석
        report.append("📰 === AI TIMES 뉴스 분석 ===")
        report.append("")
        
        categorized_news = {}
        for news in news_list:
            category = news.get('category', '기타')
            if category not in categorized_news:
                categorized_news[category] = []
            categorized_news[category].append(news)
        
        for category, articles in categorized_news.items():
            report.append(f"📌 [{category}] ({len(articles)}건)")
            report.append("-" * 80)
            
            for i, article in enumerate(articles[:5], 1):  # 카테고리당 5개만
                report.append(f"{i}. {article['title']}")
                report.append(f"   🔗 {article['url']}")
                report.append(f"   📅 {article.get('date', 'N/A')}")
                
                # LLM 분석 (선택적)
                if analyze_news and i <= 2:  # 각 카테고리 상위 2개만 분석
                    print(f"🤖 LLM 분석 중: {article['title'][:50]}...")
                    
                    instruction = """이 뉴스를 다음 형식으로 분석해주세요:
1. 핵심 내용 (2-3문장, 초등학생 수준)
2. 왜 중요한가? (1-2문장)
3. 당신이 해야 할 행동 (1-2개 항목)

간결하고 명확하게 작성해주세요."""
                    
                    analysis = self.analyze_with_qwen(article['title'], instruction)
                    report.append(f"\n   🤖 AI 분석:")
                    for line in analysis.split('\n'):
                        if line.strip():
                            report.append(f"      {line.strip()}")
                    
                    # 팩트 체크 (선택적)
                    fact_check = self.fact_check_with_brave(article['title'][:100])
                    if fact_check:
                        report.append(f"\n   ✅ 팩트 체크 (Brave Search):")
                        for fc in fact_check[:2]:
                            report.append(f"      • {fc['title']}")
                            report.append(f"        {fc['url']}")
                    
                    time.sleep(2)  # API 레이트 리밋 고려
                
                report.append("")
            
            report.append("")
        
        # 2. 허깅페이스 트렌딩
        report.append("🤗 === 허깅페이스 트렌딩 TOP 30 ===")
        report.append("")
        
        # 모델
        report.append("🔥 트렌딩 모델 TOP 30")
        report.append("-" * 80)
        for i, model in enumerate(self.huggingface_data['models'][:30], 1):
            report.append(f"{i:2d}. {model['name']}")
            report.append(f"    📊 다운로드: {model['downloads']:,} | ❤️ 좋아요: {model['likes']:,}")
            report.append(f"    🏷️  Task: {model['task']}")
            report.append(f"    🔗 {model['url']}")
            report.append("")
        
        report.append("")
        
        # 스페이스
        report.append("🚀 트렌딩 스페이스 TOP 30")
        report.append("-" * 80)
        for i, space in enumerate(self.huggingface_data['spaces'][:30], 1):
            report.append(f"{i:2d}. {space['name']}")
            report.append(f"    📝 {space['title']}")
            report.append(f"    🔗 {space['url']}")
            report.append("")
        
        # 3. 종합 요약
        report.append("=" * 80)
        report.append("📈 종합 요약")
        report.append("=" * 80)
        report.append(f"• 총 뉴스 수집: {len(news_list)}건")
        report.append(f"• 카테고리 수: {len(categorized_news)}개")
        report.append(f"• 트렌딩 모델: {len(self.huggingface_data['models'])}개")
        report.append(f"• 트렌딩 스페이스: {len(self.huggingface_data['spaces'])}개")
        report.append("")
        
        return '\n'.join(report)
    
    def run_full_analysis(self, news_urls: List[str], analyze_with_llm: bool = True) -> str:
        """전체 분석 실행"""
        print("🚀 AI 뉴스 & 허깅페이스 트렌딩 분석 시작...")
        print("")
        
        # 1. 뉴스 수집
        news_list = self.fetch_aitimes_news(news_urls)
        print(f"✅ 총 {len(news_list)}건의 뉴스 수집 완료")
        print("")
        
        # 2. 뉴스 카테고리 분류
        categorized_news = self.categorize_news(news_list)
        print("✅ 뉴스 카테고리 분류 완료")
        print("")
        
        # 3. 허깅페이스 트렌딩 수집
        self.fetch_huggingface_trending()
        print("")
        
        # 4. 리포트 생성
        print("📝 리포트 생성 중...")
        report = self.generate_report(categorized_news, analyze_news=analyze_with_llm)
        
        print("")
        print("✅ 분석 완료!")
        
        return report
    
    def save_report(self, report: str, filename: str = None):
        """리포트 저장"""
        if filename is None:
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            filename = f"ai_news_report_{timestamp}.txt"
        
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(report)
        
        print(f"💾 리포트 저장 완료: {filename}")


# ==================== 사용 예시 ====================

def main():
    """메인 실행 함수"""
    
    # API 키 설정
    FIREWORKS_API_KEY = "YOUR_FIREWORKS_API_KEY"  # 여기에 Fireworks API 키 입력
    BRAVE_API_KEY = "YOUR_BRAVE_API_KEY"  # 여기에 Brave Search API 키 입력
    
    # AI Times 뉴스 URL
    news_urls = [
        "https://www.aitimes.com/news/articleList.html?sc_multi_code=S2&view_type=sm",  # AI 산업
        "https://www.aitimes.com/news/articleList.html?sc_section_code=S1N24&view_type=sm"  # AI 기술
    ]
    
    # 분석기 초기화
    analyzer = AINewsAnalyzer(
        fireworks_api_key=FIREWORKS_API_KEY,
        brave_api_key=BRAVE_API_KEY
    )
    
    # 전체 분석 실행
    # analyze_with_llm=False로 설정하면 LLM 분석 없이 빠르게 수집만 함
    report = analyzer.run_full_analysis(
        news_urls=news_urls,
        analyze_with_llm=True  # LLM 분석 활성화 (시간이 오래 걸림)
    )
    
    # 결과 출력
    print("\n" + "=" * 80)
    print(report)
    
    # 파일 저장
    analyzer.save_report(report)


if __name__ == "__main__":
    main()


# ==================== 사용 팁 ====================
"""
1. API 키 설정:
   - Fireworks AI: https://fireworks.ai/
   - Brave Search: https://brave.com/search/api/

2. 빠른 테스트 (LLM 분석 없이):
   analyzer.run_full_analysis(news_urls, analyze_with_llm=False)

3. 특정 카테고리만 분석:
   categorized_news에서 원하는 카테고리 필터링

4. 크롤링 주기 조정:
   time.sleep() 값을 조정하여 속도/안정성 균형

5. 결과 활용:
   - JSON으로 저장: json.dumps(analyzer.huggingface_data)
   - 데이터베이스 저장
   - 대시보드 연동
"""