GET /index.html HTTP/1.1
Host: www.example.com
User-Agent: Mozilla/5.0
Accept: text/html

# Python中检查状态码
import requests

response = requests.get('https://httpbin.org/status/200')
if response.status_code == 200:
    print('请求成功！')
elif response.status_code == 403:
    print('被拒绝了，需要加请求头或使用代理')
elif response.status_code == 404:
    print('页面不存在')

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                  '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Referer': 'https://www.google.com/',
}
response = requests.get('https://example.com', headers=headers)

import requests

# 创建一个 Session 对象，可以自动保存 Cookie
session = requests.Session()

# 模拟登录，提交表单
login_data = {'username': 'user', 'password': 'pass'}
session.post('https://example.com/login', data=login_data)

# 后续请求会自动带上登录后的 Cookie
response = session.get('https://example.com/dashboard')
print(response.text)

import requests

# GET 请求
resp = requests.get('https://api.github.com')
print(resp.json())  # 自动解析 JSON

# POST 请求
resp = requests.post('https://httpbin.org/post', data={'key': 'value'})

# 带参数的 GET 请求
params = {'q': 'python', 'page': 1}
resp = requests.get('https://httpbin.org/get', params=params)

import httpx

# 同步方式（和 requests 类似）
resp = httpx.get('https://httpbin.org/get')
print(resp.status_code)

# 异步方式（需要 async/await）
async def fetch():
    async with httpx.AsyncClient() as client:
        resp = await client.get('https://httpbin.org/get')
        return resp.text

# 支持 HTTP/2
resp = httpx.get('https://httpbin.org/get', http2=True)

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://httpbin.org/get')
        print(html)

asyncio.run(main())

import re
import requests

html = requests.get('https://httpbin.org/html').text

# 提取标题
title_pattern = r'
(.*?)
'
match = re.search(title_pattern, html)
if match:
    print('标题:', match.group(1))

# 提取所有链接
link_pattern = r'href=["\'](https?://[^"\']+)["\']'
links = re.findall(link_pattern, html)
print('找到链接:', links[:5])

from bs4 import BeautifulSoup
import requests

html = requests.get('https://httpbin.org/html').text
soup = BeautifulSoup(html, 'html.parser')

# 通过标签名获取
h1 = soup.find('h1')
print('标题:', h1.text)

# 通过 class 查找
items = soup.find_all(class_='item')
for item in items:
    print(item.text)

# 通过属性查找
link = soup.find('a', href=True)
print('链接:', link['href'])

# CSS 选择器（通过 select 方法）
titles = soup.select('div.content h2')
for title in titles:
    print(title.text.strip())

from lxml import html
import requests

resp = requests.get('https://httpbin.org/html')
tree = html.fromstring(resp.content)

# XPath 定位
title = tree.xpath('//h1/text()')
print('标题:', title[0] if title else '')

# 通过 class 定位
items = tree.xpath('//div[@class="item"]')
for item in items:
    print(item.text)

# 提取属性值
links = tree.xpath('//a/@href')
print('所有链接:', links[:5])

# 多条件组合
result = tree.xpath('//div[@id="main"]//a[contains(@href, "article")]/text()')
print(result)

import requests
from bs4 import BeautifulSoup

resp = requests.get('https://example.com')
soup = BeautifulSoup(resp.text, 'html.parser')
# 如果这里找不到内容，但浏览器打开能看到，说明是动态渲染
print(soup.prettify()[:1000])

import requests

# 直接调用 API 获取数据，比渲染页面高效得多
api_url = 'https://example.com/api/articles?page=1'
headers = {
    'User-Agent': 'Mozilla/5.0 ...',
    'Accept': 'application/json',
    'X-Requested-With': 'XMLHttpRequest',
}
resp = requests.get(api_url, headers=headers)
data = resp.json()  # 直接拿到结构化的 JSON 数据
print(data)

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
import time

# 配置无头浏览器（不显示界面）
options = Options()
options.add_argument('--headless')  # 无头模式
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('User-Agent=Mozilla/5.0 ...')

driver = webdriver.Chrome(options=options)

# 访问页面
driver.get('https://example.com')

# 等待 JavaScript 渲染完成
time.sleep(3)  # 简单粗暴
# 或者使用显式等待
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
wait = WebDriverWait(driver, 10)
element = wait.until(EC.presence_of_element_located((By.CLASS_NAME, 'content')))

# 获取渲染后的页面源码
html = driver.page_source

# 传给 BeautifulSoup 解析
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
print(soup.find('h1').text)

driver.quit()

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    
    # 拦截网络请求（可选）
    page.route('**/*.{png,jpg,jpeg,gif}', lambda route: route.abort())
    
    # 访问页面
    page.goto('https://example.com')
    
    # 等待某个元素出现
    page.wait_for_selector('.content', timeout=10000)
    
    # 获取渲染后的内容
    content = page.content()
    
    # 点击和交互
    page.click('button#load-more')
    page.wait_for_timeout(2000)
    
    # 截图（调试用）
    page.screenshot(path='debug.png')
    
    browser.close()

const puppeteer = require('puppeteer');

(async () => {
    const browser = await puppeteer.launch({ headless: true });
    const page = await browser.newPage();
    await page.goto('https://example.com');
    await page.waitForSelector('.content');
    const html = await page.content();
    console.log(html);
    await browser.close();
})();

import csv

data = [
    {'title': '文章一', 'url': 'https://example.com/1', 'date': '2025-01-01'},
    {'title': '文章二', 'url': 'https://example.com/2', 'date': '2025-01-02'},
]

with open('articles.csv', 'w', newline='', encoding='utf-8-sig') as f:
    writer = csv.DictWriter(f, fieldnames=['title', 'url', 'date'])
    writer.writeheader()
    writer.writerows(data)

print('已保存到 articles.csv')

import json

data = {
    'source': 'example.com',
    'count': 2,
    'articles': [
        {'title': '文章一', 'url': 'https://example.com/1'},
        {'title': '文章二', 'url': 'https://example.com/2'},
    ]
}

with open('articles.json', 'w', encoding='utf-8') as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

print('已保存到 articles.json')

import sqlite3

conn = sqlite3.connect('crawler.db')
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,
        content TEXT,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# 插入数据
cursor.execute(
    'INSERT OR IGNORE INTO articles (title, url, content) VALUES (?, ?, ?)',
    ('文章标题', 'https://example.com/1', '文章内容...')
)
conn.commit()

# 查询数据
cursor.execute('SELECT * FROM articles LIMIT 10')
for row in cursor.fetchall():
    print(row)

conn.close()

import pymysql

conn = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    database='crawler',
    charset='utf8mb4',
)

cursor = conn.cursor()
cursor.execute('''
    CREATE TABLE IF NOT EXISTS articles (
        id INT AUTO_INCREMENT PRIMARY KEY,
        title VARCHAR(500) NOT NULL,
        url VARCHAR(2000) UNIQUE NOT NULL,
        content LONGTEXT,
        crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
''')

# 批量插入
sql = 'INSERT IGNORE INTO articles (title, url) VALUES (%s, %s)'
data = [('文章一', 'https://example.com/1'), ('文章二', 'https://example.com/2')]
cursor.executemany(sql, data)
conn.commit()

conn.close()

"""
Hacker News 热门文章爬虫
功能：抓取首页文章列表，提取标题、链接、分数，存入 SQLite
"""

import requests
from bs4 import BeautifulSoup
import sqlite3
import time
import re


def setup_database():
    """创建数据库和表"""
    conn = sqlite3.connect('hackernews.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS stories (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            url TEXT,
            score INTEGER DEFAULT 0,
            author TEXT,
            comments INTEGER DEFAULT 0,
            crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    return conn


def fetch_page(url):
    """发送 HTTP 请求，获取页面内容"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                      'AppleWebKit/537.36 (KHTML, like Gecko) '
                      'Chrome/120.0.0.0 Safari/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.5',
    }

    print(f'正在请求: {url}')
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()  # 如果状态码不是 200，抛出异常
        return response.text
    except requests.RequestException as e:
        print(f'请求失败: {e}')
        return None


def parse_stories(html):
    """解析 HTML，提取文章信息"""
    soup = BeautifulSoup(html, 'html.parser')
    stories = []

    # Hacker News 每一行是一个 
    for row in soup.select('tr.athing'):
        title_cell = row.find('td', class_='title')
        if not title_cell:
            continue

        # 标题和链接
        link_tag = title_cell.find('a')
        if not link_tag:
            continue

        title = link_tag.get_text(strip=True)
        url = link_tag.get('href', '')

        # 如果是相对路径，拼接完整 URL
        if url.startswith('item?'):
            url = f'https://news.ycombinator.com/{url}'

        # 获取分数和评论数（在下一行  中）
        subtext_row = row.find_next_sibling('tr')
        score = 0
        author = ''
        comments = 0

        if subtext_row:
            subtext = subtext_row.find('td', class_='subtext')
            if subtext:
                score_span = subtext.find('span', class_='score')
                if score_span:
                    score_text = score_span.get_text()
                    score = int(re.search(r'\d+', score_text).group())

                author_tag = subtext.find('a', class_='hnuser')
                if author_tag:
                    author = author_tag.get_text(strip=True)

                # 评论数
                links = subtext.find_all('a')
                for link in links:
                    text = link.get_text()
                    if 'comment' in text:
                        num = re.search(r'\d+', text)
                        if num:
                            comments = int(num.group())

        stories.append({
            'title': title,
            'url': url,
            'score': score,
            'author': author,
            'comments': comments,
        })

    return stories


def save_to_database(conn, stories):
    """将文章数据存入数据库"""
    cursor = conn.cursor()
    inserted = 0

    for story in stories:
        try:
            cursor.execute('''
                INSERT OR IGNORE INTO stories (title, url, score, author, comments)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                story['title'],
                story['url'],
                story['score'],
                story['author'],
                story['comments'],
            ))
            if cursor.rowcount > 0:
                inserted += 1
        except sqlite3.Error as e:
            print(f'数据库写入错误: {e}')

    conn.commit()
    return inserted


def main():
    """主函数：执行爬虫流程"""
    print('=' * 50)
    print('Hacker News 热门文章爬虫')
    print('=' * 50)

    # 1. 初始化数据库
    conn = setup_database()

    # 2. 获取页面
    html = fetch_page('https://news.ycombinator.com/')
    if not html:
        conn.close()
        return

    # 3. 解析数据
    stories = parse_stories(html)
    print(f'解析到 {len(stories)} 篇文章')

    # 4. 存储数据
    count = save_to_database(conn, stories)
    print(f'新增了 {count} 条记录')

    # 5. 展示结果
    print('\n抓取结果预览:')
    print('-' * 50)
    for i, story in enumerate(stories[:5], 1):
        print(f'{i}. [{story["score"]}分] {story["title"]}')
        print(f'   作者: {story["author"]} | 评论: {story["comments"]}')
        print(f'   链接: {story["url"][:70]}...')
        print()

    # 6. 关闭数据库连接
    conn.close()
    print(f'数据已保存到 hackernews.db')
    print('爬取完成！')


if __name__ == '__main__':
    main()

==================================================
Hacker News 热门文章爬虫
==================================================
正在请求: https://news.ycombinator.com/
解析到 30 篇文章
新增了 30 条记录

抓取结果预览:
--------------------------------------------------
1. [423分] Show HN: I built a CLI tool to generate landing pages
   作者: user123 | 评论: 158
   链接: https://example.com/cli-landing-page...
2. [387分] Ask HN: What are you working on?
   作者: techfan | 评论: 342
   ...
数据已保存到 hackernews.db
爬取完成！

# 查看网站的 robots.txt
import requests

resp = requests.get('https://example.com/robots.txt')
print(resp.text)

User-agent: *
Disallow: /admin/
Disallow: /private/
Allow: /public/

Sitemap: https://example.com/sitemap.xml

import re
import requests

def check_robots_txt(base_url):
    """检查 robots.txt，返回是否允许爬取"""
    robots_url = base_url.rstrip('/') + '/robots.txt'
    try:
        resp = requests.get(robots_url, timeout=5)
        if resp.status_code == 200:
            print(f'robots.txt 内容:\n{resp.text}')
            # 这里可以用 urllib.robotparser 做更完整的解析
            from urllib.robotparser import RobotFileParser
            rp = RobotFileParser()
            rp.parse(resp.text.splitlines())
            can_fetch = rp.can_fetch('MyCrawler/1.0', base_url)
            print(f'允许爬取首页: {can_fetch}')
            return rp
    except Exception as e:
        print(f'无法读取 robots.txt: {e}')
    return None

import time
import random

class PoliteCrawler:
    """有礼貌的爬虫——控制请求频率"""

    def __init__(self, min_delay=1.0, max_delay=3.0):
        self.min_delay = min_delay
        self.max_delay = max_delay
        self.last_request_time = 0

    def wait_before_request(self):
        """在每次请求之前等待"""
        elapsed = time.time() - self.last_request_time
        delay = random.uniform(self.min_delay, self.max_delay)
        if elapsed < delay:
            sleep_time = delay - elapsed
            print(f'等待 {sleep_time:.1f} 秒...')
            time.sleep(sleep_time)
        self.last_request_time = time.time()

    def get(self, url, **kwargs):
        self.wait_before_request()
        print(f'请求: {url}')
        return requests.get(url, **kwargs)


# 使用示例
crawler = PoliteCrawler(min_delay=1.5, max_delay=3.0)
for page in range(1, 6):
    resp = crawler.get(f'https://example.com/articles?page={page}')
    # 处理响应...

# ❌ 不好的做法
headers = {'User-Agent': 'python-requests/2.31.0'}

# ✅ 好的做法
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                  'AppleWebKit/537.36 (KHTML, like Gecko) '
                  'Chrome/120.0.0.0 Safari/537.36'
}

import random

USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ...',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ...',
]

def get_random_headers():
    return {
        'User-Agent': random.choice(USER_AGENTS),
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    }

import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_resilient_session(retries=3, backoff_factor=0.5):
    """创建一个带有重试机制的 Session"""
    session = requests.Session()

    retry_strategy = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[500, 502, 503, 504],
    )

    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount('http://', adapter)
    session.mount('https://', adapter)

    return session


# 使用
session = create_resilient_session()
try:
    resp = session.get('https://example.com', timeout=10)
    resp.raise_for_status()
except requests.RequestException as e:
    print(f'多次重试后仍然失败: {e}')