Skip to content

Development

Jump to bottom
Justin Gu edited this page May 1, 2026 · 1 revision

开发与贡献指南

面向想要参与开发或二次开发的贡献者。

项目架构

app/
├── __init__.py          # create_app() 应用工厂
├── extensions.py        # 扩展单例（db, jwt, csrf, socketio...）
├── api/                 # REST API 蓝图（JWT + Session 认证）
│   ├── __init__.py      # 注册所有 API 蓝图到 /api 前缀
│   ├── auth.py          # Session 认证（/api/auth/）
│   ├── auth_jwt.py      # JWT 认证（/api/auth/jwt/）
│   ├── xiaomi_auth.py   # 小米账号绑定（/api/xiaomi/）
│   ├── devices.py       # 设备管理（/api/devices/）
│   ├── homes.py         # 家庭管理（/api/homes/）
│   ├── scenes.py        # 场景执行（/api/scenes/）
│   ├── tokens.py        # API Token（/api/tokens/）
│   ├── device_groups.py # 设备分组（/api/groups/）
│   ├── automations.py   # 自动化规则（/api/automations/）
│   └── energy.py        # 能耗统计（/api/energy/）
├── web/                 # Web UI 蓝图（Session + CSRF 认证）
│   ├── routes.py        # 前端页面路由
│   ├── admin.py         # 管理后台路由
│   └── socketio.py      # SocketIO 事件处理
├── services/            # 业务逻辑层（不操作 HTTP request/response）
├── models/              # SQLAlchemy 数据模型
├── schemas/             # Marshmallow 序列化/校验
├── utils/               # 工具类
│   ├── response.py      # 统一响应函数 success() / error()
│   ├── decorators.py    # auth_required 等装饰器
│   └── mijia_pool.py    # MijiaAPIAdapter（从 DB 初始化）
└── cli/                 # Flask CLI 集成
config/                  # Flask 配置类
migrations/              # Alembic 数据库迁移
tests/                   # 测试

核心设计原则

分层架构

请求 → API/Web 路由 → Service → Model/外部 API → 响应
  • 路由层（api/、web/）：只做参数校验和响应构造
  • 服务层（services/）：业务逻辑，不直接操作 request/response
  • 模型层（models/）：数据库操作

统一响应

所有 API 使用 app/utils/response.py 的辅助函数：

from app.utils.response import success, error

# 成功响应
return success(data={"key": "value"}, message="操作成功")

# 错误响应
return error("参数错误", 400)

响应格式：

{"success": true, "data": {}, "message": "ok"}
{"success": false, "message": "错误描述"}

认证

接口类型 认证方式 装饰器
REST API JWT 或 Session @auth_required
Web 页面 Session + CSRF @login_required

@auth_required 同时支持 JWT Bearer Token 和 Session Cookie。

MijiaAPIAdapter

位于 app/utils/mijia_pool.py，从数据库中的 auth_data 直接初始化，不走文件 I/O。

# 正确用法 — 从数据库初始化
adapter = MijiaAPIAdapter(user_id)
result = adapter.get_prop(did, prop_name)

# 错误 — 不要改为文件路径方式

开发环境搭建

# 1. 克隆并安装
git clone https://github.com/handsomejustin/mijia-control.git
cd mijia-control
python -m venv venv && source venv/bin/activate
pip install -e ".[dev]"

# 2. 配置 .env（参见 Installation 页面）
cp .env.example .env

# 3. 初始化数据库
mysql -u root -p -e "CREATE DATABASE mijia CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
flask db upgrade

# 4. 启动开发服务器
python run.py

代码规范

使用 Ruff 进行 lint 和格式化。

ruff check .              # 检查问题
ruff check --fix .        # 自动修复
ruff format .             # 格式化

配置（pyproject.toml）：

  • line-length = 120
  • target-version = "py39"
  • 规则：E, F, W, I

测试

# 运行所有测试
pytest -v

# 仅运行 service 层测试
pytest tests/test_services/ -v

# 查看覆盖率
pytest --cov=app --cov-report=term-missing -v

测试数据库使用 TEST_DATABASE_URL（参见 .env.example），默认用 SQLite。

数据库迁移

# 生成迁移脚本（修改 Model 后执行）
flask db migrate -m "add new table"

# 应用迁移
flask db upgrade

# 回滚
flask db downgrade

添加新 API 模块的步骤

1. 创建 Model

在 app/models/ 下创建数据模型：

# app/models/my_model.py
from app.extensions import db

class MyModel(db.Model):
    __tablename__ = "my_table"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), nullable=False)

2. 创建 Schema

在 app/schemas/ 下创建序列化/校验 schema：

# app/schemas/my_schema.py
from marshmallow import Schema, fields, validate

class MySchema(Schema):
    name = fields.Str(required=True, validate=validate.Length(min=1, max=64))

3. 创建 Service

在 app/services/ 下编写业务逻辑：

# app/services/my_service.py
from app.extensions import db
from app.models.my_model import MyModel

class MyService:
    @staticmethod
    def list_items(user_id):
        return MyModel.query.filter_by(user_id=user_id).all()

4. 创建 API 路由

在 app/api/ 下创建蓝图：

# app/api/my_module.py
from flask import Blueprint, request
from app.utils.decorators import auth_required, get_current_user_id
from app.utils.response import error, success

my_ns = Blueprint("my_module", __name__, url_prefix="/my-module")

@my_ns.route("/", methods=["GET"])
@auth_required
def list_items():
    from app.services.my_service import MyService
    items = MyService.list_items(get_current_user_id())
    return success(data=items)

5. 注册蓝图

在 app/api/__init__.py 中注册：

from app.api.my_module import my_ns
api_bp.register_blueprint(my_ns)

6. 生成迁移并测试

flask db migrate -m "add my_module"
flask db upgrade
pytest -v

贡献流程

  1. Fork 仓库
  2. 创建功能分支：git checkout -b feature/my-feature
  3. 编写代码和测试
  4. 确保通过 lint 和测试： 
    ruff check .
pytest -v
  5. 提交代码：git commit -m "feat: 简短描述"
  6. 推送分支：git push origin feature/my-feature
  7. 创建 Pull Request

Commit 规范

<type>: <description>

类型：
feat     新功能
fix      修复 Bug
refactor 重构（不改变功能）
docs     文档更新
test     测试相关
chore    构建/工具/依赖变更
perf     性能优化

注意事项

  • mijiaAPI 的 execute_text_directive 的 quiet 参数需要转为 int
  • SocketIO 使用 async_mode="threading"，不是 eventlet 异步模式
  • 安全头在 create_app() 的 _add_security_headers() 中全局设置
  • CSRF 对 API 请求豁免（JWT 不需要 CSRF），但 Web 表单需要
  • 限流默认 200/day、50/hour

下一步：FAQ — 常见问题与排错。

Clone this wiki locally