dealerhub/backend/scripts/check_migrations.py

"""迁移健康检查（独立脚本，需在真实库上运行）。

为什么不是 pytest 用例：pytest-django 的测试库按**模型**建表并绕过迁移状态，
`MigrationAutodetector` 在 pytest 进程里返回空——"模型改了没生成迁移"这类问题
在单元测试里**测不出来**（实测过，会假通过）。

本脚本用真实 settings 跑，等价于 `makemigrations --check` + `migrate --check`，
但额外做几件 pytest 做不到的事：
1. 检查模型字段与**真实库**列是否一致（能抓到"迁移生成了但没 migrate"）
2. 检查新 app 的表是否真存在
3. 退出码非 0，可直接接 CI / 部署前钩子

用法：
    python scripts/check_migrations.py            # 用 DJANGO_SETTINGS_MODULE（默认 dev）
    DJANGO_SETTINGS_MODULE=config.settings.prod python scripts/check_migrations.py
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

# 让脚本能 import 到项目（backend/ 为根）
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")

import django  # noqa: E402

django.setup()

from django.apps import apps as django_apps  # noqa: E402
from django.core.management import call_command  # noqa: E402
from django.db import connection  # noqa: E402
from io import StringIO  # noqa: E402


OWN_APPS = [
    "core", "catalog", "partner", "inventory", "finance",
    "purchase", "sales", "report", "channel", "notify",
    "openapi", "printing", "ai", "billing", "website", "storefront",
]

RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
RESET = "\033[0m"


def ok(msg: str):
    print(f"  {GREEN}✓{RESET} {msg}")


def bad(msg: str):
    print(f"  {RED}✗{RESET} {msg}")


def warn(msg: str):
    print(f"  {YELLOW}!{RESET} {msg}")


def check_pending_migrations() -> int:
    """1. 模型改了但没 makemigrations。"""
    print("\n[1] 待生成的迁移（makemigrations --check）")
    out = StringIO()
    try:
        call_command("makemigrations", "--check", "--dry-run",
                     stdout=out, stderr=out, verbosity=1)
        ok("无待生成迁移")
        return 0
    except SystemExit as exc:
        if exc.code == 0:
            ok("无待生成迁移")
            return 0
        text = out.getvalue().strip()
        bad("存在未生成的迁移：")
        for line in text.splitlines():
            print("      " + line)
        print(f"\n    修复：python manage.py makemigrations")
        return 1


def check_unapplied_migrations() -> int:
    """2. 有迁移文件但没执行 migrate。"""
    print("\n[2] 未应用的迁移（migrate --check）")
    out = StringIO()
    try:
        call_command("migrate", "--check", stdout=out, stderr=out, verbosity=0)
        ok("所有迁移已应用")
        return 0
    except SystemExit as exc:
        if exc.code == 0:
            ok("所有迁移已应用")
            return 0
        text = out.getvalue().strip()
        bad("存在未应用的迁移：")
        for line in text.splitlines()[:20]:
            print("      " + line)
        print(f"\n    修复：python manage.py migrate")
        return 1


def check_model_columns() -> int:
    """3. 模型字段 vs 真实库列（能抓到 schema 漂移）。"""
    print("\n[3] 模型字段与数据库列一致性")
    failures = []
    checked = 0
    # 引擎无关：用 Django 的 introspection API（SQLite/PG/MySQL 都支持）
    with connection.cursor() as cur:
        for app_label in OWN_APPS:
            try:
                config = django_apps.get_app_config(app_label)
            except LookupError:
                continue
            for model in config.get_models():
                table = model._meta.db_table
                try:
                    desc = connection.introspection.get_table_description(cur, table)
                except Exception:
                    failures.append(f"{app_label}.{model.__name__}: 表 {table} 不存在")
                    continue
                cols = {c.name for c in desc}
                checked += 1
                for field in model._meta.concrete_fields:
                    if field.column not in cols:
                        failures.append(
                            f"{app_label}.{model.__name__}: 缺列 {field.column}"
                        )
    if failures:
        bad(f"{len(failures)} 处不一致：")
        for f in failures[:30]:
            print("      " + f)
        print("\n    修复：python manage.py makemigrations && python manage.py migrate")
        return 1
    ok(f"{checked} 个模型的字段与数据库一致")
    return 0


def check_new_app_tables() -> int:
    """4. 关键新表存在性（新 app 最容易漏迁移）。"""
    print("\n[4] 关键表存在性")
    required = [
        ("ai", "AiUsage"),
        ("billing", "Plan"),
        ("billing", "Subscription"),
        ("storefront", "StorefrontAccount"),
        ("storefront", "CustomerProductAuth"),
        ("storefront", "StorefrontOrder"),
        ("storefront", "StorefrontOrderLine"),
    ]
    missing = []
    for app_label, model_name in required:
        try:
            model = django_apps.get_model(app_label, model_name)
        except LookupError:
            missing.append(f"{app_label}.{model_name} 模型未注册")
            continue
        try:
            model.objects.exists()
        except Exception as exc:
            missing.append(f"{app_label}.{model_name}: {type(exc).__name__}")
    if missing:
        bad("以下关键表不可用：")
        for m in missing:
            print("      " + m)
        return 1
    ok(f"{len(required)} 张关键表可查")
    return 0


def main() -> int:
    db = connection.settings_dict
    print(f"迁移健康检查 · 库={db.get('NAME')} · 引擎={db.get('ENGINE')}")
    failures = 0
    failures += check_pending_migrations()
    failures += check_unapplied_migrations()
    failures += check_model_columns()
    failures += check_new_app_tables()

    print()
    if failures:
        print(f"{RED}检查未通过：{failures} 项问题{RESET}")
        return 1
    print(f"{GREEN}全部通过{RESET}")
    return 0


if __name__ == "__main__":
    sys.exit(main())