Python 全栈 + 自动化 + 双数据库

Django · HTML/CSS/JS · MySQL · MongoDB · 爬虫 · 办公自动化 · 全函数级实战手册

🗺️ 全栈项目开发顺序(黄金流程)

涉及8大技术栈的项目,必须按"需求→数据→后端→前端→采集→自动化→部署"的顺序推进:

Step 1: 需求与数据建模

先确定访问模式,再决定用MySQL还是MongoDB,画出ER图/文档结构。

Step 2: Django项目骨架

创建项目/App、配置双数据库路由、定义Model/Schema。

Step 3: API/视图层开发

编写Views/Serializers,用DRF或原生View暴露接口。

Step 4: 前端页面实现

HTML语义化→CSS响应式布局→JS交互对接API。

Step 5: 数据采集补充

爬虫抓取外部数据写入MySQL/MongoDB,补全数据源。

Step 6: 办公自动化串联

定时导出报表、生成Word/Excel、邮件通知等业务闭环。

Step 7: 测试+部署+监控

单元测试、E2E测试、Docker容器化、Nginx反代、日志告警。

⚠️ 黄金法则:数据模型定生死。不要边写代码边改表结构,先花30%时间把数据层设计清楚。

🐍 Python 核心函数速查表

1. 内置高频函数

函数用途示例
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函数结果缓存递归/重复计算加速

2. 异步与并发

# 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()

🎸 Django 全函数详解

1. Model 层核心函数

方法说明典型用法
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()事务块保证数据一致性

2. View 层核心装饰器与类

# 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')

3. DRF 序列化器核心

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

4. Django 双数据库路由配置

# 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  # 简化示例

🌐 HTML/CSS/JS 前端核心方法

1. CSS 现代布局函数

属性/函数用途关键值
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)

2. JS DOM 与 Fetch 核心

// 事件委托(性能最优)
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' });

🐬 MySQL 全操作指南

1. Python 操作 MySQL (PyMySQL / mysqlclient)

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)

2. 索引与优化核心命令

命令/概念用途注意
EXPLAIN ANALYZE查看真实执行计划+耗时MySQL 8.0+
CREATE INDEX idx_xxx创建索引遵循最左前缀原则
Covering Index覆盖索引避免回表SELECT字段都在索引中
SHOW PROCESSLIST查看当前连接/慢查询排查锁等待
pt-online-schema-change在线DDL不锁表大表改结构必备

🍃 MongoDB 全操作指南

1. PyMongo 核心操作

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))

2. 索引设计 ESR 原则

Equality(等值) → Sort(排序) → Range(范围)
例: find({status:"A"}).sort({date:-1}).filter({price:{$gt:100}})
✅ 正确索引: {status:1, date:-1, price:1}

🕷️ 爬虫全方法详解

1. 技术选型矩阵

场景推荐方案核心函数
静态页面/APIrequests + bs4/lxmlget/post, select/xpath
JS渲染/SPAPlaywrightgoto, wait_for_selector, evaluate
大规模分布式Scrapy + RedisSpider, Pipeline, Middleware
APP抓包mitmproxy + Frida中间人代理、Hook
验证码识别ddddocr / 打码平台classification, detection

2. Playwright 核心方法

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()
🚫 合规红线:遵守robots.txt、控制频率(≤2req/s)、不爬隐私数据、商用前法务审核。

⚙️ 办公自动化全函数库

1. Excel (openpyxl / pandas)

任务核心函数
数据分析/清洗pandasread_excel, groupby, merge, to_excel
样式/图表/模板openpyxlload_workbook, Font/Fill/Alignment, Chart
高性能写入XlsxWriteradd_format, write_formula, add_chart
读取超大文件openpyxlread_only=True 流式读取

2. Word (python-docx / docxtpl)

# 模板渲染(推荐)
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")

3. PDF / 邮件 / 调度

任务核心函数
PDF文本提取pdfplumberextract_text(), extract_tables()
PDF合并/拆分PyPDF2merge(), extract_page()
发送邮件smtplib + emailMIMEMultipart, SMTP_SSL
定时任务APSchedulerCronTrigger, IntervalTrigger
GUI自动化pyautoguilocateOnScreen, click, typewrite

🔗 全技术栈整合架构图

┌─────────────┐     ┌──────────────────────────────────────────┐     ┌─────────────┐
│  前端 SPA    │◄───►│           Django Backend                 │◄───►│  外部API/网站 │
│ HTML/CSS/JS │     │  ┌─────────┐  ┌──────────┐  ┌─────────┐ │     │  (爬虫采集)  │
└─────────────┘     │  │ Views/  │  │ Celery   │  │ DRF API │ │     └─────────────┘
                    │  │ Templates│  │ Worker   │  │         │ │
                    │  └────┬────┘  └────┬─────┘  └────┬────┘ │
                    │       │            │              │      │
                    │  ┌────▼────┐  ┌────▼─────┐  ┌────▼────┐ │
                    │  │ MySQL   │  │ Redis    │  │ MongoDB │ │
                    │  │(结构化) │  │(缓存/队列)│  │(文档/日志)│ │
                    │  └─────────┘  └──────────┘  └─────────┘ │
                    └──────────────────┬───────────────────────┘
                                       │ Celery Beat / Cron
                              ┌────────▼────────┐
                              │  办公自动化任务   │
                              │ Excel/Word/PDF  │
                              │ 邮件/企微通知    │
                              └─────────────────┘
💡 整合要点:
① MySQL存核心业务数据(用户/订单),MongoDB存非结构化数据(爬虫结果/日志/分析)
② 爬虫结果通过Celery异步写入,不阻塞Web请求
③ 办公自动化任务由Celery Beat定时触发,与Web共享Model层
④ 前端通过DRF API交互,Django Template仅用于管理后台/简单页面
⑤ 所有敏感配置走环境变量,数据库密码绝不硬编码