python采集数据的几种方式

import requestsimport jsonimport time# 1. 查找API# 方法：浏览器开发者工具 -> Network -> XHR# 看请求的URL和参数，很多网站是GET请求带参数的def fetch_from_api(api_url, params=None, headers=None):    """通用API数据获取函数"""    try:        response = requests.get(            api_url,            params=params,            headers=headers or {'User-Agent': 'AI-Research/1.0'}        )        if response.status_code == 200:            return response.json()        else:            print(f"API请求失败: {response.status_code}")            return None    except Exception as e:        print(f"请求出错: {e}")        return None# 示例：调用GitHub API获取项目信息github_api = "https://api.github.com/search/repositories"params = {    'q': 'machine learning',    'sort': 'stars',    'order': 'desc',    'per_page': 10}# data = fetch_from_api(github_api, params)# if data:#     for repo in data['items']:#         print(f"{repo['name']}: {repo['description']}")

import requestsfrom bs4 import BeautifulSoupimport pandas as pdfrom urllib.parse import urljoinclass ResearchDataCollector:    """研究用数据采集器"""    def __init__(self, base_url):        self.base_url = base_url        self.session = requests.Session()        self.session.headers.update({            'User-Agent': 'Mozilla/5.0 (AI-Research-Project/1.0; +http://your-research-site.edu)'        })        self.collected_data = []    def fetch_page(self, url):        """获取单页内容"""        try:            response = self.session.get(url, timeout=10)            response.raise_for_status()            return response.text        except requests.RequestException as e:            print(f"获取页面失败 {url}: {e}")            return None    def extract_data(self, html, selectors):        """根据选择器提取数据"""        soup = BeautifulSoup(html, 'html.parser')        data = {}        # 示例：提取学术论文信息        for key, selector in selectors.items():            element = soup.select_one(selector)            if element:                data[key] = element.get_text(strip=True)            else:                data[key] = None        return data    def crawl_pages(self, start_url, max_pages=10):        """爬取多页数据"""        current_url = start_url        page_count = 0        while current_url and page_count < max_pages:            print(f"采集第 {page_count + 1} 页: {current_url}")            html = self.fetch_page(current_url)            if html:                # 提取数据（这里需要根据实际网页结构调整）                data = self.extract_data(html, {                    'title': 'h1.article-title',                    'authors': '.author-list',                    'abstract': '.abstract-content',                    'date': '.publish-date'                })                if data['title']:  # 确保有实际内容                    self.collected_data.append(data)            # 找下一页链接（实际需要根据网站结构调整）            # 这里只是示例逻辑            page_count += 1            if page_count < max_pages:                current_url = self.get_next_page(html)            time.sleep(2)  # 礼貌性延迟        return self.collected_data    def get_next_page(self, html):        """获取下一页链接"""        soup = BeautifulSoup(html, 'html.parser')        next_link = soup.find('a', text='下一页')        if next_link and 'href' in next_link.attrs:            return urljoin(self.base_url, next_link['href'])        return None    def save_data(self, filename):        """保存采集的数据"""        df = pd.DataFrame(self.collected_data)        df.to_csv(filename, index=False, encoding='utf-8-sig')        print(f"数据已保存到 {filename}，共 {len(df)} 条记录")# 使用示例# collector = ResearchDataCollector("https://arxiv.org")# data = collector.crawl_pages("https://arxiv.org/list/cs.AI/recent", max_pages=3)# collector.save_data("ai_papers.csv")

from selenium import webdriverfrom selenium.webdriver.common.by import Byfrom selenium.webdriver.support.ui import WebDriverWaitfrom selenium.webdriver.support import expected_conditions as ECimport jsonclass DynamicDataCollector:    """动态网站数据采集"""    def __init__(self):        # 使用Headless模式，不显示浏览器窗口        options = webdriver.ChromeOptions()        options.add_argument('--headless')        options.add_argument('--no-sandbox')        options.add_argument('--disable-dev-shm-usage')        # 添加研究用User-Agent        options.add_argument('user-agent=AI-Research-Bot/1.0 (+https://research.example.edu)')        self.driver = webdriver.Chrome(options=options)        self.wait = WebDriverWait(self.driver, 10)    def collect_scroll_data(self, url, scroll_times=5):        """采集需要滚动加载的数据"""        self.driver.get(url)        all_data = []        for i in range(scroll_times):            # 等待内容加载            time.sleep(2)            # 提取当前可见内容            items = self.driver.find_elements(By.CSS_SELECTOR, '.data-item')  # 根据实际调整            for item in items:                try:                    data = {                        'text': item.text,                        'timestamp': item.get_attribute('data-time'),                        # 根据需要添加更多字段                    }                    all_data.append(data)                except:                    continue            # 模拟滚动            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")            print(f"已滚动 {i+1} 次，收集到 {len(all_data)} 条数据")            time.sleep(1)  # 滚动后等待        return all_data    def close(self):        self.driver.quit()# 使用示例# collector = DynamicDataCollector()# tweets = collector.collect_scroll_data("https://twitter.com/search?q=machine%20learning", 3)# collector.close()

import scrapyclass ResearchSpider(scrapy.Spider):    name = 'research_spider'    custom_settings = {        'USER_AGENT': 'AI-Research-Project/1.0 (+http://research.example.edu)',        'DOWNLOAD_DELAY': 2,  # 礼貌延迟        'CONCURRENT_REQUESTS': 1,    }    def start_requests(self):        # 起始URL        urls = ['https://example.com/data']        for url in urls:            yield scrapy.Request(url=url, callback=self.parse)    def parse(self, response):        # 解析逻辑        items = response.css('.data-item')        for item in items:            yield {                'title': item.css('h2::text').get(),                'content': item.css('.content::text').get(),                'url': response.url,            }

import pandas as pdimport redef clean_research_data(df):    """清洗研究数据"""    # 1. 去重    df = df.drop_duplicates(subset=['title', 'content'])    # 2. 处理缺失值    df = df.dropna(subset=['content'])    # 3. 清理文本    df['clean_content'] = df['content'].apply(lambda x: re.sub(r'\s+', ' ', str(x)).strip())    # 4. 过滤无效数据    df = df[df['clean_content'].str.len() > 50]  # 至少50个字符    # 5. 标准化格式    if 'date' in df.columns:        df['date'] = pd.to_datetime(df['date'], errors='coerce')    return df# 保存处理后的数据# df_cleaned = clean_research_data(df)# df_cleaned.to_json('cleaned_data.json', orient='records', force_ascii=False)