Ubuntu
428 lines
15 KiB
Python
428 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
国内技术网站抓取器
|
|
支持掘金、OSChina、知乎热榜
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import sqlite3
|
|
import datetime
|
|
import time
|
|
import random
|
|
import logging
|
|
import re
|
|
import html
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional
|
|
import requests
|
|
|
|
# 配置日志
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class JuejinCrawler:
|
|
"""掘金抓取器 - 使用官方API"""
|
|
|
|
API_URL = "https://api.juejin.cn/recommend_api/v1/article/recommend_all_feed"
|
|
|
|
def fetch_hot_articles(self, limit: int = 10) -> List[Dict]:
|
|
"""获取热门文章"""
|
|
articles = []
|
|
|
|
try:
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
|
|
}
|
|
|
|
payload = {
|
|
"cursor": "0",
|
|
"id_type": 2,
|
|
"limit": limit,
|
|
"sort_type": 200 # 200=热门
|
|
}
|
|
|
|
response = requests.post(
|
|
self.API_URL,
|
|
headers=headers,
|
|
json=payload,
|
|
timeout=15
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
|
|
if data.get('data'):
|
|
for item in data['data']:
|
|
article_info = item.get('item_info', {}).get('article_info', {})
|
|
author_info = item.get('item_info', {}).get('author_user_info', {})
|
|
|
|
title = article_info.get('title', '')
|
|
article_id = article_info.get('article_id', '')
|
|
brief = article_info.get('brief_content', '')
|
|
|
|
if title and article_id:
|
|
articles.append({
|
|
'title': title,
|
|
'url': f"https://juejin.cn/post/{article_id}",
|
|
'description': brief[:200] if brief else '',
|
|
'source': '掘金',
|
|
'category': 'tech',
|
|
'author': author_info.get('user_name', ''),
|
|
'word_count': article_info.get('word_count', 0)
|
|
})
|
|
|
|
logger.info(f"掘金API获取到 {len(articles)} 篇文章")
|
|
else:
|
|
logger.warning(f"掘金API返回空数据: {data.get('err_msg', '未知错误')}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"掘金API请求失败: {e}")
|
|
|
|
return articles
|
|
|
|
|
|
class OSChinaCrawler:
|
|
"""OSChina抓取器 - 使用AJAX API"""
|
|
|
|
API_URL = "https://www.oschina.net/action/ajax/get_more_recommend_blog"
|
|
|
|
def fetch_hot_articles(self, limit: int = 10) -> List[Dict]:
|
|
"""获取热门博客文章"""
|
|
articles = []
|
|
|
|
try:
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
|
'X-Requested-With': 'XMLHttpRequest',
|
|
'Referer': 'https://www.oschina.net/blog'
|
|
}
|
|
|
|
url = f"{self.API_URL}?classification=0&p=1"
|
|
|
|
req = urllib.request.Request(url, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=15) as response:
|
|
content = response.read().decode('utf-8')
|
|
|
|
# 解析HTML内容
|
|
titles = re.findall(r'<h2 class="blog-name[^"]*"[^>]*>(.*?)</h2>', content, re.DOTALL)
|
|
links = re.findall(r'href="(https://my\.oschina\.net/[^"]+/blog/\d+)"', content)
|
|
briefs = re.findall(r'<section class="blog-brief[^"]*"[^>]*>(.*?)</section>', content, re.DOTALL)
|
|
|
|
for i, (title_html, url) in enumerate(zip(titles[:limit], links[:limit])):
|
|
title = html.unescape(re.sub(r'<[^>]+>', '', title_html)).strip()
|
|
brief = ''
|
|
if i < len(briefs):
|
|
brief = html.unescape(re.sub(r'<[^>]+>', '', briefs[i])).strip()[:200]
|
|
|
|
if title and url:
|
|
articles.append({
|
|
'title': title,
|
|
'url': url,
|
|
'description': brief,
|
|
'source': 'OSChina',
|
|
'category': 'tech',
|
|
'author': '',
|
|
'word_count': 0
|
|
})
|
|
|
|
logger.info(f"OSChina获取到 {len(articles)} 篇文章")
|
|
|
|
except Exception as e:
|
|
logger.error(f"OSChina请求失败: {e}")
|
|
|
|
return articles
|
|
|
|
|
|
class ZhihuHotCrawler:
|
|
"""知乎热榜抓取器"""
|
|
|
|
def __init__(self, cookie_file: str = "/home/ubuntu/zhihu_cookies.txt"):
|
|
self.cookie_file = cookie_file
|
|
|
|
def fetch_hot_list(self, limit: int = 10) -> List[Dict]:
|
|
"""获取知乎热榜"""
|
|
articles = []
|
|
|
|
try:
|
|
# 读取cookies（支持JSON格式）
|
|
cookies = {}
|
|
if os.path.exists(self.cookie_file):
|
|
with open(self.cookie_file, 'r') as f:
|
|
cookie_content = f.read().strip()
|
|
# 尝试解析为JSON（浏览器导出格式）
|
|
try:
|
|
cookie_list = json.loads(cookie_content)
|
|
if isinstance(cookie_list, list):
|
|
for cookie in cookie_list:
|
|
if 'name' in cookie and 'value' in cookie:
|
|
cookies[cookie['name']] = cookie['value']
|
|
elif isinstance(cookie_list, dict):
|
|
cookies = cookie_list
|
|
except json.JSONDecodeError:
|
|
# 回退到简单的key=value格式
|
|
for item in cookie_content.split(';'):
|
|
if '=' in item:
|
|
key, value = item.strip().split('=', 1)
|
|
cookies[key] = value
|
|
|
|
headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
|
'Referer': 'https://www.zhihu.com/',
|
|
}
|
|
|
|
# 知乎热榜API
|
|
url = "https://www.zhihu.com/api/v3/feed/topstory/hot-lists/total"
|
|
|
|
response = requests.get(url, headers=headers, cookies=cookies, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
|
|
if data.get('data'):
|
|
for item in data['data'][:limit]:
|
|
target = item.get('target', {})
|
|
title = target.get('title', '')
|
|
question_id = target.get('id', '')
|
|
excerpt = target.get('excerpt', '')
|
|
|
|
# 只保留技术相关问题（简单过滤）
|
|
tech_keywords = ['编程', '代码', '开发', 'AI', '人工智能', '算法',
|
|
'Python', 'Java', '前端', '后端', '数据库', '云',
|
|
'技术', '软件', '系统', '架构', '工程师', '程序员',
|
|
'ChatGPT', '大模型', '机器学习', '深度学习', '区块链']
|
|
|
|
is_tech = any(kw in title for kw in tech_keywords)
|
|
|
|
if title and question_id:
|
|
articles.append({
|
|
'title': title,
|
|
'url': f"https://www.zhihu.com/question/{question_id}",
|
|
'description': excerpt[:200] if excerpt else '',
|
|
'source': '知乎热榜',
|
|
'category': 'tech' if is_tech else 'general',
|
|
'author': '',
|
|
'word_count': 0,
|
|
'is_tech': is_tech
|
|
})
|
|
|
|
logger.info(f"知乎热榜获取到 {len(articles)} 个问题")
|
|
|
|
except Exception as e:
|
|
logger.error(f"知乎热榜请求失败: {e}")
|
|
|
|
return articles
|
|
|
|
|
|
class ChinaTechCrawler:
|
|
"""国内技术网站统一抓取器"""
|
|
|
|
def __init__(self, config_path: str = None):
|
|
if config_path is None:
|
|
config_path = "/home/ubuntu/nanocode/it_automation_enhanced/config/config.json"
|
|
|
|
self.config_path = config_path
|
|
self.load_config()
|
|
|
|
# 初始化各平台抓取器
|
|
self.juejin = JuejinCrawler()
|
|
self.oschina = OSChinaCrawler()
|
|
self.zhihu = ZhihuHotCrawler()
|
|
|
|
def load_config(self):
|
|
"""加载配置"""
|
|
try:
|
|
with open(self.config_path, 'r', encoding='utf-8') as f:
|
|
self.config = json.load(f)
|
|
except Exception as e:
|
|
logger.error(f"配置加载失败: {e}")
|
|
self.config = {}
|
|
|
|
def init_database(self) -> bool:
|
|
"""初始化数据库"""
|
|
db_path = self.config.get('paths', {}).get('database_path')
|
|
if not db_path:
|
|
db_path = "/home/ubuntu/nanocode/it_automation_enhanced/data/articles.db"
|
|
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS articles (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT NOT NULL,
|
|
url TEXT UNIQUE NOT NULL,
|
|
source TEXT NOT NULL,
|
|
category TEXT,
|
|
publish_date TEXT,
|
|
word_count INTEGER DEFAULT 0,
|
|
content TEXT,
|
|
processed INTEGER DEFAULT 0,
|
|
translation_status INTEGER DEFAULT 0,
|
|
humanizer_status INTEGER DEFAULT 0,
|
|
xhs_status INTEGER DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
''')
|
|
|
|
cursor.execute('CREATE INDEX IF NOT EXISTS idx_url ON articles(url)')
|
|
cursor.execute('CREATE INDEX IF NOT EXISTS idx_processed ON articles(processed)')
|
|
cursor.execute('CREATE INDEX IF NOT EXISTS idx_source ON articles(source)')
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
logger.info(f"数据库初始化完成: {db_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"数据库初始化失败: {e}")
|
|
return False
|
|
|
|
def save_article(self, article: Dict) -> tuple:
|
|
"""保存文章到数据库"""
|
|
db_path = self.config.get('paths', {}).get('database_path')
|
|
if not db_path:
|
|
db_path = "/home/ubuntu/nanocode/it_automation_enhanced/data/articles.db"
|
|
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
# 检查是否已存在
|
|
cursor.execute('SELECT id FROM articles WHERE url = ?', (article['url'],))
|
|
existing = cursor.fetchone()
|
|
|
|
if existing:
|
|
logger.debug(f"文章已存在，跳过: {article['title'][:50]}...")
|
|
conn.close()
|
|
return True, False
|
|
|
|
# 插入新文章
|
|
cursor.execute('''
|
|
INSERT INTO articles
|
|
(title, url, source, category, publish_date, word_count, content)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
''', (
|
|
article['title'],
|
|
article['url'],
|
|
article.get('source', 'unknown'),
|
|
article.get('category', 'tech'),
|
|
datetime.datetime.now().strftime('%Y-%m-%d'),
|
|
article.get('word_count', 0),
|
|
article.get('description', '')
|
|
))
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
logger.info(f"新文章保存成功: [{article.get('source')}] {article['title'][:50]}...")
|
|
return True, True
|
|
|
|
except Exception as e:
|
|
logger.error(f"文章保存失败: {e}")
|
|
return False, False
|
|
|
|
def fetch_all_sources(self, limit_per_source: int = 5) -> List[Dict]:
|
|
"""从所有源抓取文章"""
|
|
all_articles = []
|
|
|
|
# 获取配置中启用的源
|
|
websites = self.config.get('websites', [])
|
|
|
|
for site in websites:
|
|
if not site.get('enabled', True):
|
|
continue
|
|
|
|
source_name = site.get('name', '')
|
|
articles = []
|
|
|
|
if source_name == '掘金':
|
|
articles = self.juejin.fetch_hot_articles(limit_per_source)
|
|
elif source_name == 'OSChina':
|
|
articles = self.oschina.fetch_hot_articles(limit_per_source)
|
|
elif source_name == '知乎热榜':
|
|
articles = self.zhihu.fetch_hot_list(limit_per_source)
|
|
# 知乎只保留技术相关问题
|
|
articles = [a for a in articles if a.get('is_tech', False)]
|
|
|
|
all_articles.extend(articles)
|
|
|
|
# 避免请求过快
|
|
time.sleep(1)
|
|
|
|
return all_articles
|
|
|
|
def run(self) -> int:
|
|
"""运行抓取器"""
|
|
logger.info("=" * 60)
|
|
logger.info("🇨🇳 开始国内技术网站抓取")
|
|
logger.info("=" * 60)
|
|
|
|
# 初始化数据库
|
|
if not self.init_database():
|
|
logger.error("数据库初始化失败，退出")
|
|
return 0
|
|
|
|
# 抓取所有源
|
|
articles = self.fetch_all_sources(limit_per_source=5)
|
|
|
|
if not articles:
|
|
logger.warning("未获取到任何文章")
|
|
return 0
|
|
|
|
logger.info(f"共获取 {len(articles)} 篇文章，开始保存...")
|
|
|
|
# 保存文章
|
|
saved_count = 0
|
|
new_count = 0
|
|
|
|
for article in articles:
|
|
success, is_new = self.save_article(article)
|
|
if success:
|
|
saved_count += 1
|
|
if is_new:
|
|
new_count += 1
|
|
|
|
logger.info("=" * 60)
|
|
logger.info(f"📊 抓取完成: 处理 {saved_count} 篇，新增 {new_count} 篇")
|
|
|
|
return new_count
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
crawler = ChinaTechCrawler()
|
|
|
|
try:
|
|
saved_count = crawler.run()
|
|
|
|
if saved_count > 0:
|
|
print(f"\n🎉 抓取成功，新增 {saved_count} 篇文章")
|
|
return 0
|
|
else:
|
|
print("\n⚠️ 抓取完成，但未新增文章")
|
|
return 1
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n⏹️ 用户中断")
|
|
return 130
|
|
except Exception as e:
|
|
logger.error(f"抓取过程中出现错误: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|