installed_apps = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'channels', # 添加这行
'your_app', # 你的应用名
]

asgi_application = "your_project.asgi.application"

# channels 配置
channel_layers = {
'default': {
'backend': 'channels.layers.inmemorychannellayer'
}
}

import os
ｆrom django.core.asgi import get_asgi_application
ｆrom channels.routing import protocoltyperouter, urlrouter
ｆrom channels.auth import authmiddlewarestack
import your_app.routing # 稍后我们会创建这个

os.environ.setdefault('django_settings_module', 'your_project.settings')

application = protocoltyperouter({
"http": get_asgi_application(),
"websocket": authmiddlewarestack(
urlrouter(
your_app.routing.websocket_urlpatterns
)
),
})

ｆrom channels.generic.websocket import asyncwebsocketconsumer
import json

class chatconsumer(asyncwebsocketconsumer):
async def connect(self):
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = f'chat_{self.room_name}'

# 加入房间组
await self.channel_layer.group_add(
self.room_group_name,
self.channel_name
)

await self.accept()

async def disconnect(self, close_code):
# 离开房间组
await self.channel_layer.group_discard(
self.room_group_name,
self.channel_name
)

async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']

# 发送消息到房间组
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'chat_message',
'message': message
}
)

async def chat_message(self, event):
message = event['message']

# 发送消息到 websocket
await self.send(text_data=json.dumps({
'message': message
}))

ｆrom django.urls import re_path
ｆrom . import consumers

websocket_urlpatterns = [
re_path(r'ws/chat/(?pw+)/$', consumers.chatconsumer.as_asgi()),
]

ｆrom django.shortcuts import render

def chat_room(request, room_name):
return render(request, 'chat_room.html', {
'room_name': room_name
})

{{ room_name|json_script:"room-name" }}

ｆrom django.urls import path
ｆrom . import views

urlpatterns = [
path('chat//', views.chat_room, name='chat_room'),
]

[pytest]
django_settings_module = your_project.settings
python_files = tests.py test_*.py *_tests.py

your_app/
tests/
__init__.py
test_models.py
test_views.py

import pytest
from your_app.models import yourmodel

@pytest.mark.django_db
def test_your_model():
model = yourmodel.objects.create(name="test")
assert model.name == "test"

import pytest
from django.urls import reverse

@pytest.mark.django_db
def test_your_view(client):
url = reverse('your-view-name')
response = client.get(url)
assert response.status_code == 200

import pytest
from your_app.models import yourmodel

@pytest.fixture
def sample_model(db):
return yourmodel.objects.create(name="sample")

def test_your_model_with_fixture(sample_model):
assert sample_model.name == "sample"

test:
script:
- pip install -r requirements.txt
- pytest --cov=your_app

django-admin startproject myproject
cd myproject

from django.db import models

class item(models.model):
name = models.charfield(max_length=100)
description = models.textfield()

def __str__(self):
return self.name

python manage.py makemigrations
python manage.py migrate

from django.shortcuts import render
from .models import item

def item_list(request):
items = item.objects.all()
return render(request, 'item_list.html', {'items': items})

from django.urls import include, path

urlpatterns = [
path('', include('myapp.urls')),
]

from django.urls import path
from . import views

urlpatterns = [
path('', views.item_list, name='item_list'),
]

items

{% for item in items %}

• {{ item.name }} - {{ item.description }}

{% endfor %}

import requests
import json

def fetch_ajax_data(url, params, headers):
response = requests.get(url, params=params, headers=headers)
return response.json()

# 示例使用
url = 'https://api.example.com/data'
params = {
'page': 1,
'limit': 10
}
headers = {
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36',
'referer': 'https://www.example.com',
'x-requested-with': 'xmlhttprequest'
}

data = fetch_ajax_data(url, params, headers)
print(json.dumps(data, indent=2))

from selenium import webdriver
from selenium.webdriver.chrome.options import options
from selenium.webdriver.common.by import by
from selenium.webdriver.support.ui import webdriverwait
from selenium.webdriver.support import expected_conditions as ec
import json

def fetch_ajax_data_with_selenium(url):
chrome_options = options()
chrome_options.add_argument("--headless")

driver = webdriver.chrome(options=chrome_options)
driver.get(url)

# 等待特定元素加载完成
element = webdriverwait(driver, 10).until(
ec.presence_of_element_located((by.class_name, "data-container"))
)

# 提取数据
data = driver.ｅxecute_script("return window.ajaxdata;") # 假设数据存储在全局变量中

driver.quit()
return data

# 使用示例
url = 'https://www.example.com/ajax-page'
data = fetch_ajax_data_with_selenium(url)
print(json.dumps(data, indent=2))

import requests

def fetch_https_data(url):
response = requests.get(url)
response.raise_for_status() # 如果请求不成功则抛出异常
return response.text

# 使用示例
url = 'https://api.github.com'
data = fetch_https_data(url)
print(data)

import requests
import certifi

def fetch_https_data_with_cert(url):
response = requests.get(url, verify=certifi.where())
response.raise_for_status()
return response.text

# 使用示例
url = 'https://self-signed.badssl.com/'
try:
data = fetch_https_data_with_cert(url)
print(data)
except requests.exceptions.sslerror as e:
print(f"ssl证书验证失败: {e}")

import requests

def fetch_https_data_with_client_cert(url, cert_path):
response = requests.get(url, cert=cert_path)
response.raise_for_status()
return response.text

# 使用示例
url = 'https://client.badssl.com/'
cert_path = ('path/to/client.crt', 'path/to/client.key')
try:
data = fetch_https_data_with_client_cert(url, cert_path)
print(data)
except requests.exceptions.requestexception as e:
print(f"请求失败: {e}")

import requests
import json
from selenium import webdriver
from selenium.webdriver.chrome.options import options
from selenium.webdriver.common.by import by
from selenium.webdriver.support.ui import webdriverwait
from selenium.webdriver.support import expected_conditions as ec

class advancedscraper:
def __init__(self):
self.session = requests.session()
self.session.headers.update({
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36'
})

chrome_options = options()
chrome_options.add_argument("--headless")
self.driver = webdriver.chrome(options=chrome_options)

def fetch_api_data(self, url, params=none):
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()

def fetch_ajax_data(self, url):
self.driver.get(url)

# 等待ajax内容加载
webdriverwait(self.driver, 10).until(
ec.presence_of_element_located((by.class_name, "ajax-content"))
)

# 提取数据
data = self.driver.ｅxecute_script("return window.ajaxdata;")
return data

def scrape_website(self, base_url, api_endpoint):
# 爬取主页面
main_page = self.session.get(base_url)
main_page.raise_for_status()

# 爬取api数据
api_data = self.fetch_api_data(api_endpoint)

# 爬取ajax加载的数据
ajax_data = self.fetch_ajax_data(base_url)

return {
'main_page': main_page.text,
'api_data': api_data,
'ajax_data': ajax_data
}

def close(self):
self.driver.quit()

# 使用示例
def main():
scraper = advancedscraper()
try:
data = scraper.scrape_website(
'https://example.com',
'https://api.example.com/data'
)
print(json.dumps(data, indent=2))
finally:
scraper.close()

if __name__ == '__main__':
main()

import requests
from lxml import etree

def scrape_website(url):
# 发送http请求
headers = {
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36'
}
response = requests.get(url, headers=headers)

# 解析html
html = etree.html(response.content)

# 使用xpath提取信息
title = html.xpath('//title/text()')[0]
paragraphs = html.xpath('//p/text()')
links = html.xpath('//a/@href')

return {
'title': title,
'paragraphs': paragraphs,
'links': links
}

# 使用示例
result = scrape_website('https://example.com')
print(f"title: {result['title']}")
print(f"number of paragraphs: {len(result['paragraphs'])}")
print(f"number of links: {len(result['links'])}")

# 选择所有包含"news"类的div元素
news_divs = html.xpath('//div[contains(@class, "news")]')

# 选择第一个段落
first_paragraph = html.xpath('//p[1]/text()')

# 选择所有class为"content"且id为"main"的div元素
content_divs = html.xpath('//div[@class="content" and @id="main"]')

# 选择所有具有子元素p的div元素
divs_with_p = html.xpath('//div[child::p]')

# 选择所有具有属性class的元素的父元素
parents = html.xpath('//*[@class]/parent::*')

import requests
from lxml import etree
import csv

def scrape_news_website(url):
headers = {
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36'
}
response = requests.get(url, headers=headers)
html = etree.html(response.content)

# 提取新闻文章
articles = []
for article in html.xpath('//div[@class="article"]'):
title = article.xpath('.//h2/text()')[0]
summary = article.xpath('.//p[@class="summary"]/text()')[0]
author = article.xpath('.//span[@class="author"]/text()')[0]
date = article.xpath('.//span[@class="date"]/text()')[0]
link = article.xpath('.//a[@class="read-more"]/@href')[0]

articles.append({
'title': title,
'summary': summary,
'author': author,
'date': date,
'link': link
})

return articles

def save_to_csv(articles, filename):
with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['title', 'summary', 'author', 'date', 'link']
writer = csv.dictwriter(csvfile, fieldnames=fieldnames)

writer.writeheader()
for article in articles:
writer.writerow(article)

# 主函数
def main():
url = 'https://example-news-site.com'
articles = scrape_news_website(url)
save_to_csv(articles, 'news_articles.csv')
print(f"scraped {len(articles)} articles and saved to news_articles.csv")

if __name__ == '__main__':
main()

from selenium import webdriver
from selenium.webdriver.chrome.options import options
from lxml import etree
import time

def scrape_dynamic_website(url):
chrome_options = options()
chrome_options.add_argument("--headless") # 无头模式

driver = webdriver.chrome(options=chrome_options)
driver.get(url)

# 等待页面加载（你可能需要调整等待时间）
time.sleep(5)

# 获取渲染后的html
html_content = driver.page_source
driver.quit()

# 解析html
html = etree.html(html_content)

# 使用xpath提取信息
# ...（根据具体网站结构编写xpath）

return extracted_data

# 使用示例
data = scrape_dynamic_website('https://example-dynamic-site.com')

import urllib3

# 禁用警告（在生产环境中不建议这么做）
urllib3.disable_warnings()

# 创建 poolmanager 实例
http = urllib3.poolmanager()

# 发送get请求
response = http.request('get', 'https://api.example.com/users')

# 打印响应状态和数据
print(response.status)
print(response.data.decode('utf-8'))

import json

data = {'username': 'john', 'password': 'secret'}
encoded_data = json.dumps(data).encode('utf-8')

response = http.request(
'post',
'https://api.example.com/login',
body=encoded_data,
headers={'content-type': 'application/json'}
)

print(response.status)
print(response.data.decode('utf-8'))

from urllib3.util.retry import retry
from urllib3.util.timeout import timeout

retries = retry(total=3, backoff_factor=0.1)
timeout = timeout(connect=5.0, read=10.0)

http = urllib3.poolmanager(retries=retries, timeout=timeout)

response = http.request('get', 'https://api.example.com/users')

# 创建一个连接池
pool = urllib3.httpconnectionpool('api.example.com', maxsize=10)

# 使用连接池发送请求
response = pool.request('get', '/users')

import certifi

http = urllib3.poolmanager(
cert_reqs='cert_required',
ca_certs=certifi.where()
)

response = http.request('get', 'https://api.example.com/users')

import requests

# 发送get请求
response = requests.get('https://api.example.com/users')

# 打印响应状态和内容
print(response.status_code)
print(response.text)

# 自动解析json响应
data = response.json()
print(data)

data = {'username': 'john', 'password': 'secret'}
response = requests.post('https://api.example.com/login', json=data)

print(response.status_code)
print(response.json())

headers = {'user-agent': 'myapp/1.0'}
response = requests.get('https://api.example.com/users', headers=headers)

# 创建会话
session = requests.session()

# 登录
login_data = {'username': 'john', 'password': 'secret'}
session.post('https://api.example.com/login', json=login_data)

# 使用同一会话发送后续请求
response = session.get('https://api.example.com/dashboard')

files = {'file': open('document.pdf', 'rb')}
response = requests.post('https://api.example.com/upload', files=files)

from requests.adapters import httpadapter
from requests.packages.urllib3.util.retry import retry

retry_strategy = retry(
total=3,
backoff_factor=0.1,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = httpadapter(max_retries=retry_strategy)
session = requests.session()
session.mount('https://', adapter)
session.mount('http://', adapter)

response = session.get('https://api.example.com/users', timeout=5)

with requests.get('https://api.example.com/large-file', stream=true) as r:
r.raise_for_status()
with open('large-file.zip', 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
f.write(chunk)

import requests
from bs4 import beautifulsoup

def scrape_website(url):
headers = {
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36'
}

try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status() # 如果请求不成功则抛出异常

soup = beautifulsoup(response.text, 'html.parser')

# 提取标题
title = soup.title.string if soup.title else 'no title found'

# 提取所有段落文本
paragraphs = [p.text for p in soup.find_all('p')]

# 提取所有链接
links = [a['href'] for a in soup.find_all('a', href=true)]

return {
'title': title,
'paragraphs': paragraphs,
'links': links
}
except requests.requestexception as e:
print(f"请求错误: {e}")

return none

# 使用示例
result = scrape_website('https://example.com')
if result:
print(f"title: {result['title']}")
print(f"number of paragraphs: {len(result['paragraphs'])}")
print(f"number of links: {len(result['links'])}")

url = "https://www.example.com"
response = request.urlopen(url)
html = response.read().decode('utf-8')
print(html)

data = parse.urlencode({'key1': 'value1', 'key2': 'value2'}).encode('utf-8')
req = request.request(url, data=data, method='post')
response = request.urlopen(req)
html = response.read().decode('utf-8')
print(html)

headers = {
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36'
}
req = request.request(url, headers=headers)
response = request.urlopen(req)

print(response.getheaders())
print(response.getheader('content-type'))

encoded_url = parse.quote('https://example.com/path with spaces')
print(encoded_url)

decoded_url = parse.unquote('https://example.com/path%20with%20spaces')
print(decoded_url)

params = {'key1': 'value1', 'key2': 'value2'}
query_string = parse.urlencode(params)
url = f"https://example.com/search?{query_string}"
print(url)

try:
response = request.urlopen("https://www.example.com/nonexistent")
except error.httperror as e:
print(f"http error: {e.code}")
except error.urlerror as e:
print(f"url error: {e.reason}")

proxy_handler = request.proxyhandler({'http': 'http://127.0.0.1:8080'})
opener = request.build_opener(proxy_handler)
request.install_opener(opener)
response = request.urlopen(url)

import http.cookiejar

cookie_jar = http.cookiejar.cookiejar()
opener = request.build_opener(request.httpcookieprocessor(cookie_jar))
request.install_opener(opener)
response = request.urlopen(url)

for cookie in cookie_jar:
print(f"{cookie.name}: {cookie.value}")

class customhttphandler(request.httphandler):
def http_request(self, req):
req.add_header('custom-header', 'customvalue')
return super().http_request(req)

opener = request.build_opener(customhttphandler)
request.install_opener(opener)
response = request.urlopen(url)

class noredirecthandler(request.httpredirecthandler):
def http_error_302(self, req, fp, code, msg, headers):
return fp

opener = request.build_opener(noredirecthandler)
request.install_opener(opener)
try:
response = request.urlopen(url)
except error.httperror as e:
if e.code == 302:
print(f"redirect to: {e.headers['location']}")

import asyncio
import aiohttp

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

async def main():
urls = ['https://example.com', 'https://example.org', 'https://example.net']
tasks = [asyncio.create_task(fetch(url)) for url in urls]
results = await asyncio.gather(*tasks)
for url, html in zip(urls, results):
print(f"content length from {url}: {len(html)}")

asyncio.run(main())

from urllib import request
from bs4 import beautifulsoup

def scrape_website(url):
headers = {
'user-agent': 'mozilla/5.0 (windows nt 10.0; win64; x64) applewebkit/537.36 (khtml, like gecko) chrome/91.0.4472.124 safari/537.36'
}
req = request.request(url, headers=headers)

try:
response = request.urlopen(req)
html = response.read().decode('utf-8')

soup = beautifulsoup(html, 'html.parser')

# 提取标题
title = soup.title.string if soup.title else 'no title found'

# 提取所有段落文本
paragraphs = [p.text for p in soup.find_all('p')]

# 提取所有链接
links = [a['href'] for a in soup.find_all('a', href=true)]

return {
'title': title,
'paragraphs': paragraphs,
'links': links
}
except error.httperror as e:
print(f"http error: {e.code}")
except error.urlerror as e:
print(f"url error: {e.reason}")

return none

# 使用示例
result = scrape_website('https://example.com')
if result:
print(f"title: {result['title']}")
print(f"number of paragraphs: {len(result['paragraphs'])}")
print(f"number of links: {len(result['links'])}")

mkdir news_scraper
cd news_scraper
python -m venv venv
source venv/bin/activate # 在windows上使用 venvscriptsactivate
pip install requests beautifulsoup4 pandas lxml selenium fake-useragent

# utils.py
import time
from fake_useragent import useragent
import requests

def get_random_ua():
ua = useragent()
return ua.random

def make_request(url, max_retries=3, delay=1):
headers = {'user-agent': get_random_ua()}
for i in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
return response
except requests.requestexception as e:
print(f"请求失败 (尝试 {i+1}/{max_retries}): {e}")
if i < max_retries - 1:
time.sleep(delay)
return none

def is_valid_url(url):
try:
result = requests.urlparse(url)
return all([result.scheme, result.netloc])
except valueerror:
return false

# scraper.py
import time
from bs4 import beautifulsoup
from selenium import webdriver
from selenium.webdriver.chrome.options import options
from selenium.webdriver.common.by import by
from selenium.webdriver.support.ui import webdriverwait
from selenium.webdriver.support import expected_conditions as ec
from utils import make_request, is_valid_url

class newsscraper:
def __init__(self, base_url):
self.base_url = base_url
self.articles = []

def scrape_headlines(self):
response = make_request(self.base_url)
if not response:
print("无法获取网页内容")
return

soup = beautifulsoup(response.content, 'lxml')
headlines = soup.find_all('h2', class_='headline')

for headline in headlines:
article = {
'title': headline.text.strip(),
'url': headline.find('a')['href'] if headline.find('a') else none
}
if article['url'] and not is_valid_url(article['url']):
article['url'] = f"{self.base_url.rstrip('/')}/{article['url'].lstrip('/')}"
self.articles.append(article)

def scrape_article_content(self):
chrome_options = options()
chrome_options.add_argument("--headless")

with webdriver.chrome(options=chrome_options) as driver:
for article in self.articles:
if not article['url']:
continue

driver.get(article['url'])

try:
content = webdriverwait(driver, 10).until(
ec.presence_of_element_located((by.class_name, "article-content"))
)
article['content'] = content.text
except exception as e:
print(f"无法获取文章内容: {e}")
article['content'] = none

time.sleep(1) # 避免请求过于频繁

def get_articles(self):
return self.articles

# main.py
import pandas as pd
from scraper import newsscraper

def main():
base_url = "https://example-news-site.com"
scraper = newsscraper(base_url)

print("正在爬取新闻标题...")
scraper.scrape_headlines()

print("正在爬取文章内容...")
scraper.scrape_article_content()

articles = scraper.get_articles()

if articles:
df = pd.dataframe(articles)
df.to_csv('news_articles.csv', index=false)
print(f"已保存 {len(articles)} 篇文章到 news_articles.csv")
else:
print("未找到文章")

if __name__ == "__main__":
main()

from urllib.robotparser import robotfileparser

class newsscraper:
# ... 其他代码 ...

def check_robots_txt(self):
rp = robotfileparser()
rp.set_url(f"{self.base_url}/robots.txt")
rp.read()

if not rp.can_fetch("*", self.base_url):
print("根据robots.txt，不允许爬取此网站")
return false
return true

# 在 scrape_headlines 方法开始时调用
if not self.check_robots_txt():
return

org.springframework.boot
spring-boot-starter-data-redis

spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=your_password

@configuration
public class redisconfig {
@bean
public redistemplate redistemplate(redisconnectionfactory factory) {
redistemplate template = new redistemplate<>();
template.setconnectionfactory(factory);

// 设置key的序列化方式
template.setkeyserializer(new stringredisserializer());
// 设置value的序列化方式
template.setvalueserializer(new genericjackson2jsonredisserializer());

return template;
}
}

@service
public class userservice {
@autowired
private redistemplate redistemplate;

public void saveuser(user user) {
redistemplate.opsforvalue().set("user:" + user.getid(), user);
}

public user getuser(long id) {
return (user) redistemplate.opsforvalue().get("user:" + id);
}
}

org.springframework.boot
spring-boot-starter-cache

@enablecaching
@springbootapplication
public class application {
public static void main(string[] args) {
springapplication.run(application.class, args);
}
}

@service
public class userservice {
@cacheable(value = "users", key = "#id")
public user getuser(long id) {
// 从数据库获取用户
}

@cacheput(value = "users", key = "#user.id")
public user updateuser(user user) {
// 更新用户信息
}

@cacheevict(value = "users", key = "#id")
public void ｄeleteuser(long id) {
// 删除用户
}
}