涉及8大技术栈的项目,必须按"需求→数据→后端→前端→采集→自动化→部署"的顺序推进:
先确定访问模式,再决定用MySQL还是MongoDB,画出ER图/文档结构。
创建项目/App、配置双数据库路由、定义Model/Schema。
编写Views/Serializers,用DRF或原生View暴露接口。
HTML语义化→CSS响应式布局→JS交互对接API。
爬虫抓取外部数据写入MySQL/MongoDB,补全数据源。
定时导出报表、生成Word/Excel、邮件通知等业务闭环。
单元测试、E2E测试、Docker容器化、Nginx反代、日志告警。
| 函数 | 用途 | 示例 |
|---|---|---|
enumerate() | 带索引遍历 | for i, v in enumerate(lst): |
zip() | 并行迭代多个序列 | list(zip(keys, values)) |
map()/filter() | 函数式映射/过滤 | list(map(str.upper, names)) |
sorted() | 自定义排序 | sorted(data, key=lambda x: x['age']) |
isinstance() | 类型检查 | isinstance(val, (int, float)) |
getattr()/setattr() | 动态属性访问 | getattr(obj, field, default) |
contextlib.contextmanager | 自定义上下文管理器 | 资源安全释放 |
functools.lru_cache | 函数结果缓存 | 递归/重复计算加速 |
# asyncio 异步IO
async def fetch(session, url):
async with session.get(url) as resp:
return await resp.json()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, u) for u in urls]
results = await asyncio.gather(*tasks)
# concurrent.futures 多线程/多进程
from concurrent.futures import ThreadPoolExecutor, as_completed
with ThreadPoolExecutor(max_workers=10) as executor:
futures = {executor.submit(process, item): item for item in items}
for future in as_completed(futures):
result = future.result()
| 方法 | 说明 | 典型用法 |
|---|---|---|
objects.create() | 创建并保存 | User.objects.create(name='Alice') |
objects.get_or_create() | 获取或创建 | 防止重复插入 |
objects.bulk_create() | 批量插入 | 万级数据秒级写入 |
objects.update() | 批量更新(SQL级) | 不触发save()信号 |
select_related() | JOIN预加载(FK/O2O) | 减少N+1查询 |
prefetch_related() | 额外查询预加载(M2M) | 反向关系优化 |
annotate()/aggregate() | 分组统计 | COUNT/SUM/AVG |
values()/values_list() | 返回字典/元组 | 避免ORM对象开销 |
F() / Q() | 字段引用/复杂条件 | 原子更新、OR查询 |
transaction.atomic() | 事务块 | 保证数据一致性 |
# FBV 常用装饰器
@login_required # 登录验证
@require_POST # 限制请求方法
@csrf_exempt # 豁免CSRF(API用)
@cache_page(60*15) # 视图缓存15分钟
@permission_required('app.change_model') # 权限检查
# CBV 核心Mixin
class ArticleListView(LoginRequiredMixin, ListView):
model = Article
template_name = 'articles/list.html'
paginate_by = 20
ordering = ['-created_at']
def get_queryset(self):
return super().get_queryset().select_related('author')
class ArticleSerializer(serializers.ModelSerializer):
author_name = serializers.CharField(source='author.username', read_only=True)
tags = serializers.StringRelatedField(many=True)
class Meta:
model = Article
fields = ['id', 'title', 'content', 'author_name', 'tags', 'created_at']
extra_kwargs = {'content': {'write_only': True}}
def validate_title(self, value):
if len(value) < 5:
raise serializers.ValidationError("标题至少5个字符")
return value
# settings.py
DATABASES = {
'default': { 'ENGINE': 'django.db.backends.mysql', ... },
'mongodb': { 'ENGINE': 'djongo', 'NAME': 'mydb', ... } # 或用PyMongo直连
}
DATABASE_ROUTERS = ['myproject.routers.MultiDBRouter']
# routers.py
class MultiDBRouter:
route_app_labels = {'analytics': 'mongodb'}
def db_for_read(self, model, **hints):
if model._meta.app_label in self.route_app_labels:
return 'mongodb'
return 'default'
db_for_write = db_for_read # 简化示例
| 属性/函数 | 用途 | 关键值 |
|---|---|---|
display: grid | 二维布局 | repeat(auto-fill, minmax(280px,1fr)) |
display: flex | 一维排列 | justify-content / align-items / gap |
clamp() | 响应式尺寸 | font-size: clamp(1rem, 2.5vw, 2rem) |
:has() | 父选择器 | .card:has(img) { ... } |
container queries | 组件级响应式 | @container (min-width: 400px) |
// 事件委托(性能最优)
document.querySelector('.list').addEventListener('click', e => {
const btn = e.target.closest('[data-action]');
if (!btn) return;
const action = btn.dataset.action;
// handle action...
});
// Fetch + 错误处理封装
async function api(url, options = {}) {
const res = await fetch(url, {
headers: { 'Content-Type': 'application/json', ...options.headers },
...options
});
if (!res.ok) throw new Error(`HTTP ${res.status}: ${await res.text()}`);
return res.json();
}
// IntersectionObserver 懒加载
const observer = new IntersectionObserver(entries => {
entries.forEach(entry => {
if (entry.isIntersecting) {
entry.target.src = entry.target.dataset.src;
observer.unobserve(entry.target);
}
});
}, { rootMargin: '200px' });
import pymysql
conn = pymysql.connect(host='localhost', user='root', password='pwd',
database='mydb', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor)
with conn.cursor() as cur:
# 参数化查询(防SQL注入)
cur.execute("SELECT * FROM users WHERE age > %s AND status = %s", (18, 'active'))
rows = cur.fetchall()
# 批量插入
cur.executemany("INSERT INTO logs (msg, level) VALUES (%s, %s)", log_data)
conn.commit()
# SQLAlchemy ORM(推荐生产使用)
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session
engine = create_engine("mysql+pymysql://root:pwd@localhost/mydb")
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
| 命令/概念 | 用途 | 注意 |
|---|---|---|
EXPLAIN ANALYZE | 查看真实执行计划+耗时 | MySQL 8.0+ |
CREATE INDEX idx_xxx | 创建索引 | 遵循最左前缀原则 |
Covering Index | 覆盖索引避免回表 | SELECT字段都在索引中 |
SHOW PROCESSLIST | 查看当前连接/慢查询 | 排查锁等待 |
pt-online-schema-change | 在线DDL不锁表 | 大表改结构必备 |
from pymongo import MongoClient, ASCENDING, DESCENDING
from datetime import datetime
client = MongoClient("mongodb://localhost:27017/")
db = client["mydb"]
col = db["articles"]
# Schema Validation
db.create_collection("articles", validator={"$jsonSchema": {
"bsonType": "object",
"required": ["title", "createdAt"],
"properties": {"title": {"bsonType": "string"}}
}})
# CRUD
col.insert_one({"title": "Hello", "tags": ["py"], "createdAt": datetime.utcnow()})
col.find({"tags": "py"}).sort("createdAt", DESCENDING).limit(10)
col.update_one({"_id": oid}, {"$push": {"tags": "django"}, "$set": {"updatedAt": datetime.utcnow()}})
col.delete_many({"createdAt": {"$lt": datetime(2023,1,1)}})
# 聚合管道
pipeline = [
{"$match": {"createdAt": {"$gte": datetime(2024,1,1)}}},
{"$group": {"_id": "$tags", "count": {"$sum": 1}}},
{"$sort": {"count": -1}},
{"$limit": 10}
]
results = list(col.aggregate(pipeline))
find({status:"A"}).sort({date:-1}).filter({price:{$gt:100}}){status:1, date:-1, price:1}
| 场景 | 推荐方案 | 核心函数 |
|---|---|---|
| 静态页面/API | requests + bs4/lxml | get/post, select/xpath |
| JS渲染/SPA | Playwright | goto, wait_for_selector, evaluate |
| 大规模分布式 | Scrapy + Redis | Spider, Pipeline, Middleware |
| APP抓包 | mitmproxy + Frida | 中间人代理、Hook |
| 验证码识别 | ddddocr / 打码平台 | classification, detection |
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
ctx = browser.new_context(storage_state="auth.json") # 复用登录态
page = ctx.new_page()
page.goto("https://example.com/dashboard")
page.wait_for_selector(".data-table", state="visible")
# 拦截API响应(比解析DOM更稳)
data = []
page.on("response", lambda r: data.append(r.json()) if "/api/list" in r.url else None)
page.reload()
# 无限滚动加载
while True:
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
try: page.wait_for_selector(".load-more:not([disabled])", timeout=3000)
except: break
ctx.storage_state(path="auth.json") # 保存登录态
browser.close()
| 任务 | 库 | 核心函数 |
|---|---|---|
| 数据分析/清洗 | pandas | read_excel, groupby, merge, to_excel |
| 样式/图表/模板 | openpyxl | load_workbook, Font/Fill/Alignment, Chart |
| 高性能写入 | XlsxWriter | add_format, write_formula, add_chart |
| 读取超大文件 | openpyxl | read_only=True 流式读取 |
# 模板渲染(推荐)
from docxtpl import DocxTemplate
tpl = DocxTemplate("report_template.docx")
tpl.render({"company": "XX科技", "items": [{"name":"服务A","amt":"10万"}]})
tpl.save("output.docx")
# 从零构建
from docx import Document
doc = Document()
doc.add_heading("月度报告", level=1)
doc.add_paragraph("自动生成内容...")
table = doc.add_table(rows=3, cols=3, style="Table Grid")
doc.save("report.docx")
| 任务 | 库 | 核心函数 |
|---|---|---|
| PDF文本提取 | pdfplumber | extract_text(), extract_tables() |
| PDF合并/拆分 | PyPDF2 | merge(), extract_page() |
| 发送邮件 | smtplib + email | MIMEMultipart, SMTP_SSL |
| 定时任务 | APScheduler | CronTrigger, IntervalTrigger |
| GUI自动化 | pyautogui | locateOnScreen, click, typewrite |
┌─────────────┐ ┌──────────────────────────────────────────┐ ┌─────────────┐
│ 前端 SPA │◄───►│ Django Backend │◄───►│ 外部API/网站 │
│ HTML/CSS/JS │ │ ┌─────────┐ ┌──────────┐ ┌─────────┐ │ │ (爬虫采集) │
└─────────────┘ │ │ Views/ │ │ Celery │ │ DRF API │ │ └─────────────┘
│ │ Templates│ │ Worker │ │ │ │
│ └────┬────┘ └────┬─────┘ └────┬────┘ │
│ │ │ │ │
│ ┌────▼────┐ ┌────▼─────┐ ┌────▼────┐ │
│ │ MySQL │ │ Redis │ │ MongoDB │ │
│ │(结构化) │ │(缓存/队列)│ │(文档/日志)│ │
│ └─────────┘ └──────────┘ └─────────┘ │
└──────────────────┬───────────────────────┘
│ Celery Beat / Cron
┌────────▼────────┐
│ 办公自动化任务 │
│ Excel/Word/PDF │
│ 邮件/企微通知 │
└─────────────────┘