ForgeFlow-ERP/docs/superpowers/plans/2026-06-14-system-initialization-bootstrap.md

System Initialization Bootstrap Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Build a standard initialization tool that can create a clean ERP + miniapp database for a new customer and reset an existing test database into a clean customer-ready system.

Architecture: Add a focused backend service layer for database initialization, driven by a thin CLI script. The service imports all SQLAlchemy models, creates all ERP and miniapp tables for fresh, performs protected backups and table resets for reset, and seeds only the minimal system skeleton required for login, permissions, system config, warehouses, locations, units, and categories.

Tech Stack: Python 3.12, SQLAlchemy 2, PyMySQL, Pydantic settings, existing FastAPI backend models/services, MySQL 8, unittest/pytest-style backend tests.

---

File Structure

• Create: backend/scripts/system_initialize.py
  • CLI entry point. Parses fresh and reset, validates flags, calls the service, prints report paths.
• Create: backend/app/services/system_initializer.py
  • Core initialization service. Owns model imports, table groups, backup/report helpers, schema creation, table reset, seed routines.
• Modify: backend/app/models/miniapp.py
  • Add missing miniapp backend tables so ERP metadata can create the complete miniapp schema.
• Modify: backend/app/models/__init__.py
  • Export/import complete model modules so Base.metadata is complete in initializer and tests.
• Create: backend/tests/test_system_initializer_models.py
  • Verifies all required ERP and miniapp table names are registered in Base.metadata.
• Create: backend/tests/test_system_initializer_seed.py
  • Verifies minimal system skeleton is seeded correctly in an in-memory database.
• Create: backend/tests/test_system_initializer_reset.py
  • Verifies reset clears business data, clears people/accounts, preserves/rebuilds skeleton, and refuses destructive reset without confirmation.
• Modify: docs/superpowers/specs/2026-06-14-system-initialization-bootstrap-design.md
  • Add notes if implementation discovers a schema gap that must be reflected in design.

---

Task 1: Register Complete Miniapp Table Coverage

Files:

• Modify: backend/app/models/miniapp.py

• Modify: backend/app/models/__init__.py

• Test: backend/tests/test_system_initializer_models.py

• Step 1: Write failing metadata coverage test

Create backend/tests/test_system_initializer_models.py:

from __future__ import annotations

import unittest

from sqlalchemy import BigInteger, create_engine
from sqlalchemy.ext.compiler import compiles


@compiles(BigInteger, "sqlite")
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
    _ = type_, compiler, kw
    return "INTEGER"


import app.models.document_archive  # noqa: E402,F401
import app.models.master_data  # noqa: E402,F401
import app.models.miniapp  # noqa: E402,F401
import app.models.operations  # noqa: E402,F401
import app.models.org  # noqa: E402,F401
import app.models.planning  # noqa: E402,F401
import app.models.sales  # noqa: E402,F401
from app.models.base import Base  # noqa: E402


class SystemInitializerModelCoverageTest(unittest.TestCase):
    def test_metadata_contains_required_erp_and_miniapp_tables(self) -> None:
        required_tables = {
            "sys_department",
            "hr_employee",
            "sys_role",
            "sys_permission",
            "sys_role_permission",
            "sys_user",
            "sys_user_role",
            "sys_system_config",
            "sys_ai_assistant_config",
            "md_unit",
            "md_item_category",
            "md_item",
            "md_material",
            "md_product",
            "md_bom",
            "md_bom_item",
            "wh_warehouse",
            "wh_location",
            "wh_stock_lot",
            "wh_stock_balance",
            "wh_inventory_txn",
            "pp_production_batch_ledger",
            "pp_production_batch_ledger_txn",
            "document_archives",
            "attendance_points",
            "personnel",
            "person_roles",
            "person_attendance_points",
            "products",
            "equipment",
            "work_schedules",
            "work_sessions",
            "work_session_devices",
            "production_reports",
            "production_report_items",
            "report_audit_logs",
            "mold_lock_feedbacks",
            "device_qrcodes",
            "device_qrcode_batch_tasks",
            "notices",
            "notice_points",
            "reconciliation_ledger_entries",
        }
        self.assertTrue(required_tables.issubset(set(Base.metadata.tables)))

    def test_metadata_can_create_all_tables_on_sqlite(self) -> None:
        engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
        Base.metadata.create_all(engine)
        created_tables = set(Base.metadata.tables)
        self.assertIn("work_sessions", created_tables)
        self.assertIn("production_report_items", created_tables)
        self.assertIn("pp_production_batch_ledger", created_tables)


if __name__ == "__main__":
    unittest.main()

• Step 2: Run test to verify current failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_models.py -q

Expected: FAIL because work_sessions, work_session_devices, report_audit_logs, equipment, work_schedules, mold_lock_feedbacks, device_qrcodes, device_qrcode_batch_tasks, and reconciliation_ledger_entries are not registered in ERP metadata.

• Step 3: Add missing miniapp model classes

Modify backend/app/models/miniapp.py.

Keep existing classes, then add missing classes aligned with /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/models.py. Also expand existing fields that the miniapp backend already uses.

Add imports:

from sqlalchemy import BigInteger, Boolean, Date, DateTime, DECIMAL, ForeignKey, Integer, String, Text, text
from sqlalchemy.dialects.mysql import JSON
from sqlalchemy.orm import Mapped, mapped_column

Update MiniAppAttendancePoint to include schedule columns:

    day_start: Mapped[str] = mapped_column(String(5), server_default=text("'08:00'"))
    day_end: Mapped[str] = mapped_column(String(5), server_default=text("'17:20'"))
    lunch_start: Mapped[str] = mapped_column(String(5), server_default=text("'11:40'"))
    lunch_end: Mapped[str] = mapped_column(String(5), server_default=text("'12:40'"))
    dinner_start: Mapped[str] = mapped_column(String(5), server_default=text("'17:20'"))
    dinner_end: Mapped[str] = mapped_column(String(5), server_default=text("'18:00'"))
    overtime_start: Mapped[str] = mapped_column(String(5), server_default=text("'18:00'"))
    overtime_end: Mapped[str] = mapped_column(String(5), server_default=text("'20:00'"))
    night_start: Mapped[str] = mapped_column(String(5), server_default=text("'20:00'"))
    night_end: Mapped[str] = mapped_column(String(5), server_default=text("'06:00'"))

Add missing classes:

class MiniAppEquipment(Base):
    __tablename__ = "equipment"

    attendance_point_name: Mapped[str] = mapped_column(String(128), primary_key=True, server_default=text("''"))
    device_no: Mapped[str] = mapped_column(String(64), primary_key=True)
    device_type: Mapped[str] = mapped_column(String(32), server_default=text("'冲压设备'"))
    remark: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class MiniAppWorkSchedule(Base):
    __tablename__ = "work_schedules"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    day_start: Mapped[str] = mapped_column(String(5), server_default=text("'08:00'"))
    day_end: Mapped[str] = mapped_column(String(5), server_default=text("'17:20'"))
    lunch_start: Mapped[str] = mapped_column(String(5), server_default=text("'11:40'"))
    lunch_end: Mapped[str] = mapped_column(String(5), server_default=text("'12:40'"))
    dinner_start: Mapped[str] = mapped_column(String(5), server_default=text("'17:20'"))
    dinner_end: Mapped[str] = mapped_column(String(5), server_default=text("'18:00'"))
    overtime_start: Mapped[str] = mapped_column(String(5), server_default=text("'18:00'"))
    overtime_end: Mapped[str] = mapped_column(String(5), server_default=text("'20:00'"))
    night_start: Mapped[str] = mapped_column(String(5), server_default=text("'20:00'"))
    night_end: Mapped[str] = mapped_column(String(5), server_default=text("'06:00'"))
    attendance_latitude: Mapped[float | None] = mapped_column(DECIMAL(10, 7), nullable=True)
    attendance_longitude: Mapped[float | None] = mapped_column(DECIMAL(10, 7), nullable=True)
    attendance_radius_meters: Mapped[int] = mapped_column(Integer, server_default=text("500"))
    auto_submit_hours: Mapped[float] = mapped_column(DECIMAL(10, 2), server_default=text("15"))
    updated_by: Mapped[str | None] = mapped_column(String(20), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class MiniAppWorkSession(Base):
    __tablename__ = "work_sessions"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    attendance_point_name: Mapped[str] = mapped_column(String(128), server_default=text("''"))
    employee_phone: Mapped[str] = mapped_column(ForeignKey("personnel.phone"))
    start_at: Mapped[object] = mapped_column(DateTime)
    end_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    status: Mapped[str] = mapped_column(String(32), server_default=text("'active'"))
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class MiniAppWorkSessionDevice(Base):
    __tablename__ = "work_session_devices"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    session_id: Mapped[int] = mapped_column(ForeignKey("work_sessions.id"))
    attendance_point_name: Mapped[str] = mapped_column(String(128), server_default=text("''"))
    device_no: Mapped[str] = mapped_column(String(255))
    process_name: Mapped[str] = mapped_column(String(128), server_default=text("''"))
    scanned_at: Mapped[object] = mapped_column(DateTime)
    sort_order: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    released_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    released_by: Mapped[str | None] = mapped_column(String(20), nullable=True)
    release_reason: Mapped[str | None] = mapped_column(String(120), nullable=True)


class MiniAppReportAuditLog(Base):
    __tablename__ = "report_audit_logs"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    report_id: Mapped[int] = mapped_column(ForeignKey("production_reports.id"))
    reviewer_phone: Mapped[str] = mapped_column(ForeignKey("personnel.phone"))
    action: Mapped[str] = mapped_column(String(64))
    before_json: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    after_json: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    remark: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)


class MiniAppMoldLockFeedback(Base):
    __tablename__ = "mold_lock_feedbacks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    attendance_point_name: Mapped[str] = mapped_column(String(128), server_default=text("''"))
    mold_name: Mapped[str] = mapped_column(String(255))
    process_name: Mapped[str] = mapped_column(String(128), server_default=text("''"))
    session_device_id: Mapped[int] = mapped_column(ForeignKey("work_session_devices.id"))
    reporter_phone: Mapped[str] = mapped_column(ForeignKey("personnel.phone"))
    occupied_phone: Mapped[str] = mapped_column(ForeignKey("personnel.phone"))
    status: Mapped[str] = mapped_column(String(32), server_default=text("'pending'"))
    read_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    handled_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    handled_by: Mapped[str | None] = mapped_column(ForeignKey("personnel.phone"), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)


class MiniAppDeviceQRCode(Base):
    __tablename__ = "device_qrcodes"

    attendance_point_name: Mapped[str] = mapped_column(String(128), primary_key=True, server_default=text("''"))
    device_no: Mapped[str] = mapped_column(String(255), primary_key=True)
    process_name: Mapped[str] = mapped_column(String(128), primary_key=True, server_default=text("''"))
    qr_scene: Mapped[str] = mapped_column(String(128))
    qr_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_by: Mapped[str] = mapped_column(ForeignKey("personnel.phone"))
    created_at: Mapped[object] = mapped_column(DateTime)


class MiniAppDeviceQrBatchTask(Base):
    __tablename__ = "device_qrcode_batch_tasks"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    file_name: Mapped[str] = mapped_column(String(255))
    status: Mapped[str] = mapped_column(String(32), server_default=text("'pending'"))
    item_count: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    completed_count: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    failed_count: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    items_json: Mapped[list] = mapped_column(JSON)
    zip_url: Mapped[str | None] = mapped_column(String(500), nullable=True)
    error_message: Mapped[str | None] = mapped_column(String(1000), nullable=True)
    created_by: Mapped[str] = mapped_column(ForeignKey("personnel.phone"))
    started_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    finished_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class MiniAppReconciliationLedgerEntry(Base):
    __tablename__ = "reconciliation_ledger_entries"

    attendance_point_name: Mapped[str] = mapped_column(String(128), primary_key=True, server_default=text("''"))
    year: Mapped[int] = mapped_column(Integer, primary_key=True)
    month: Mapped[int] = mapped_column(Integer, primary_key=True)
    product_name: Mapped[str] = mapped_column(String(255), primary_key=True)
    reconciled_good_qty: Mapped[float] = mapped_column(DECIMAL(14, 2), server_default=text("0"))
    return_qty: Mapped[float] = mapped_column(DECIMAL(14, 2), server_default=text("0"))
    updated_by: Mapped[str | None] = mapped_column(String(20), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)

Expand existing classes:

• MiniAppProduct: add process_unit_price_yuan.

• MiniAppProductionReport: add review_remark, is_system_auto_submitted, auto_submit_reason, is_multi_person_assistant, multi_person_source_report_id, is_voided, voided_at, voided_by, unvoid_deadline_at.

• MiniAppProductionReportItem: add operator_count, process_unit_price_yuan, remark.

• Step 4: Update model package exports

Modify backend/app/models/__init__.py to import all model modules for side effects:

from app.models import document_archive as document_archive
from app.models import master_data as master_data
from app.models import miniapp as miniapp
from app.models import operations as operations
from app.models import org as org
from app.models import planning as planning
from app.models import sales as sales

Keep the existing class exports below those imports.

• Step 5: Run model coverage test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_models.py -q

Expected: PASS.

• Step 6: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/models/miniapp.py backend/app/models/__init__.py backend/tests/test_system_initializer_models.py
git commit -m "feat: register full miniapp schema for initialization"

---

Task 2: Add Initializer Service Constants And Options

Files:

• Create: backend/app/services/system_initializer.py

• Test: backend/tests/test_system_initializer_seed.py

• Step 1: Write failing table group test

Create backend/tests/test_system_initializer_seed.py with the first test:

from __future__ import annotations

import unittest


class SystemInitializerSeedTest(unittest.TestCase):
    def test_table_groups_have_no_duplicates_and_include_expected_tables(self) -> None:
        from app.services.system_initializer import (
            BUSINESS_RESET_TABLES,
            MINIAPP_RESET_TABLES,
            SKELETON_TABLES,
        )

        all_tables = BUSINESS_RESET_TABLES + MINIAPP_RESET_TABLES + SKELETON_TABLES
        self.assertEqual(len(all_tables), len(set(all_tables)))
        self.assertIn("so_sales_order", BUSINESS_RESET_TABLES)
        self.assertIn("wh_stock_lot", BUSINESS_RESET_TABLES)
        self.assertIn("pp_production_batch_ledger", BUSINESS_RESET_TABLES)
        self.assertIn("production_reports", MINIAPP_RESET_TABLES)
        self.assertIn("work_sessions", MINIAPP_RESET_TABLES)
        self.assertIn("sys_permission", SKELETON_TABLES)
        self.assertIn("wh_warehouse", SKELETON_TABLES)


if __name__ == "__main__":
    unittest.main()

• Step 2: Run test to verify it fails

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_seed.py::SystemInitializerSeedTest::test_table_groups_have_no_duplicates_and_include_expected_tables -q

Expected: FAIL because app.services.system_initializer does not exist.

• Step 3: Create service constants and option dataclasses

Create backend/app/services/system_initializer.py:

from __future__ import annotations

import json
import shutil
import subprocess
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable

from sqlalchemy import create_engine, inspect, select, text
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session

from app.core.config import Settings, get_settings
from app.models.base import Base
from app.services.auth import hash_password


BUSINESS_RESET_TABLES = [
    "document_archives",
    "rt_return_disposition",
    "rt_return_item",
    "rt_return_order",
    "so_delivery_item",
    "so_delivery",
    "pp_production_batch_ledger_txn",
    "pp_production_batch_ledger",
    "pp_completion_receipt_item",
    "pp_completion_receipt",
    "pp_scrap_record",
    "pp_operation_report",
    "pp_material_issue_item",
    "pp_material_issue",
    "pp_work_order_material_issue",
    "pp_work_order_operation",
    "pp_work_order_material",
    "pp_work_order",
    "wh_special_adjustment_line",
    "wh_special_adjustment",
    "wh_stocktake_adjustment",
    "wh_stocktake_line",
    "wh_stocktake_warehouse",
    "wh_stocktake",
    "wh_inventory_txn",
    "wh_stock_balance",
    "wh_stock_lot",
    "po_purchase_return_item",
    "po_purchase_return",
    "po_receipt_item",
    "po_receipt",
    "po_purchase_order_sales_order",
    "po_purchase_order_item",
    "po_purchase_order",
    "mrp_material_demand",
    "so_sales_order_item",
    "so_sales_order",
    "fi_statement_snapshot",
    "fi_cost_allocation",
    "fi_overhead_entry",
    "fi_accounting_period",
    "em_maintenance_order",
    "em_maintenance_plan",
    "em_equipment",
    "md_bom_item",
    "md_bom",
    "md_process_route_operation",
    "md_process_route",
    "md_work_center",
    "md_process",
    "md_product",
    "md_material",
    "md_item",
    "md_customer",
    "md_supplier",
    "sys_broadcast_message",
]

MINIAPP_RESET_TABLES = [
    "notice_points",
    "notices",
    "device_qrcode_batch_tasks",
    "device_qrcodes",
    "mold_lock_feedbacks",
    "report_audit_logs",
    "production_report_items",
    "production_reports",
    "work_session_devices",
    "work_sessions",
    "reconciliation_ledger_entries",
    "products",
    "equipment",
    "person_attendance_points",
    "person_roles",
    "personnel",
    "attendance_points",
    "work_schedules",
]

ACCOUNT_RESET_TABLES = [
    "sys_user_role",
    "sys_user",
    "sys_org_manager_binding",
    "sys_org_employee_binding",
    "hr_employee",
    "sys_department",
]

SKELETON_TABLES = [
    "sys_permission",
    "sys_role",
    "sys_role_permission",
    "sys_system_config",
    "sys_ai_assistant_config",
    "md_unit",
    "md_item_category",
    "wh_warehouse",
    "wh_location",
]

WAREHOUSE_TYPES = [
    ("RAW", "原材料库", "原料主库位"),
    ("SEMI", "半成品库", "半成品主库位"),
    ("FINISHED", "成品库", "成品主库位"),
    ("AUX", "辅料库", "辅料主库位"),
    ("SCRAP", "废料库", "废料主库位"),
    ("RETURN", "退货库", "退货主库位"),
]


@dataclass(frozen=True)
class SystemInitializeOptions:
    mode: str
    company_name: str
    admin_name: str
    admin_phone: str
    admin_password: str
    smart_operation_report_enabled: bool = True
    confirm_reset: bool = False
    delete_files: bool = False
    allow_any_database: bool = False


@dataclass(frozen=True)
class SystemInitializeResult:
    mode: str
    database: str
    backup_path: str | None
    summary_path: str | None
    counts_before: dict[str, int]
    counts_after: dict[str, int]
    seeded: dict[str, int | str | bool]

• Step 4: Run table group test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_seed.py::SystemInitializerSeedTest::test_table_groups_have_no_duplicates_and_include_expected_tables -q

Expected: PASS.

• Step 5: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/services/system_initializer.py backend/tests/test_system_initializer_seed.py
git commit -m "feat: define system initialization table groups"

---

Task 3: Implement Skeleton Seeding

Files:

• Modify: backend/app/services/system_initializer.py

• Modify: backend/tests/test_system_initializer_seed.py

• Step 1: Add failing skeleton seed test

Append to backend/tests/test_system_initializer_seed.py:

from datetime import datetime

from sqlalchemy import BigInteger, create_engine, select
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import Session, sessionmaker


@compiles(BigInteger, "sqlite")
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
    _ = type_, compiler, kw
    return "INTEGER"


import app.models.document_archive  # noqa: E402,F401
import app.models.master_data  # noqa: E402,F401
import app.models.miniapp  # noqa: E402,F401
import app.models.operations  # noqa: E402,F401
import app.models.org  # noqa: E402,F401
import app.models.planning  # noqa: E402,F401
import app.models.sales  # noqa: E402,F401
from app.models.base import Base  # noqa: E402
from app.models.master_data import Warehouse  # noqa: E402
from app.models.operations import WarehouseLocation  # noqa: E402
from app.models.org import Department, Employee, Permission, Role, SystemConfig, User, UserRole  # noqa: E402
from app.services.auth import verify_password  # noqa: E402


class SystemInitializerSkeletonSeedTest(unittest.TestCase):
    def setUp(self) -> None:
        engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
        Base.metadata.create_all(engine)
        self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
        self.db: Session = self.SessionLocal()

    def tearDown(self) -> None:
        self.db.close()

    def test_seed_system_skeleton_creates_admin_permissions_config_and_warehouses(self) -> None:
        from app.services.system_initializer import SystemInitializeOptions, seed_system_skeleton

        options = SystemInitializeOptions(
            mode="reset",
            company_name="百华",
            admin_name="超级管理员",
            admin_phone="13800000000",
            admin_password="secret123",
            smart_operation_report_enabled=False,
            confirm_reset=True,
        )

        result = seed_system_skeleton(self.db, options)
        self.db.commit()

        admin_user = self.db.scalar(select(User).where(User.username == "13800000000"))
        self.assertIsNotNone(admin_user)
        self.assertEqual(admin_user.is_super_admin, 1)
        self.assertTrue(verify_password("secret123", admin_user.password_hash))
        admin_employee = self.db.scalar(select(Employee).where(Employee.mobile == "13800000000"))
        self.assertEqual(admin_employee.employee_name, "超级管理员")
        root = self.db.scalar(select(Department).where(Department.dept_code == "ORG_ROOT"))
        self.assertEqual(root.dept_name, "百华")
        self.assertGreater(self.db.query(Permission).count(), 0)
        self.assertGreater(self.db.query(Role).count(), 0)
        self.assertGreater(self.db.query(UserRole).count(), 0)
        self.assertEqual(self.db.query(Warehouse).count(), 6)
        self.assertEqual(self.db.query(WarehouseLocation).count(), 6)
        prefix = self.db.scalar(select(SystemConfig).where(SystemConfig.config_code == "RAW_MATERIAL_LOT_PREFIX"))
        self.assertEqual(prefix.config_value, "YL")
        smart = self.db.scalar(select(SystemConfig).where(SystemConfig.config_code == "SMART_OPERATION_REPORT_ENABLED"))
        self.assertEqual(smart.config_value, "关闭")
        self.assertEqual(result["warehouse_count"], 6)

• Step 2: Run skeleton seed test to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_seed.py::SystemInitializerSkeletonSeedTest::test_seed_system_skeleton_creates_admin_permissions_config_and_warehouses -q

Expected: FAIL because seed_system_skeleton does not exist.

• Step 3: Implement seed_system_skeleton

Append to backend/app/services/system_initializer.py:

def _now() -> datetime:
    return datetime.now().replace(microsecond=0)


def _bool_to_config_value(enabled: bool) -> str:
    return "开启" if enabled else "关闭"


def _upsert_system_config(
    db: Session,
    *,
    code: str,
    name: str,
    value: str,
    remark: str,
) -> None:
    from app.models.org import SystemConfig

    now = _now()
    row = db.scalar(select(SystemConfig).where(SystemConfig.config_code == code))
    if row is None:
        row = SystemConfig(
            config_code=code,
            config_name=name,
            config_value=value,
            remark=remark,
            status="ACTIVE",
            created_by=None,
            updated_by=None,
            created_at=now,
            updated_at=now,
        )
        db.add(row)
        return
    row.config_name = name
    row.config_value = value
    row.remark = remark
    row.status = "ACTIVE"
    row.updated_at = now
    db.add(row)


def _upsert_role(db: Session, role_code: str, role_name: str, remark: str) -> object:
    from app.models.org import Role

    now = _now()
    row = db.scalar(select(Role).where(Role.role_code == role_code))
    if row is None:
        row = Role(
            role_code=role_code,
            role_name=role_name,
            role_scope="SYSTEM",
            status="ACTIVE",
            remark=remark,
            created_at=now,
            updated_at=now,
        )
        db.add(row)
        db.flush()
        return row
    row.role_name = role_name
    row.role_scope = "SYSTEM"
    row.status = "ACTIVE"
    row.remark = remark
    row.updated_at = now
    db.add(row)
    db.flush()
    return row


def _seed_permissions_and_roles(db: Session) -> dict[str, int]:
    from app.models.org import Permission, RolePermission
    from app.services.system_permissions import MENU_PERMISSION_TREE

    now = _now()
    roles = [
        _upsert_role(db, "ADMIN", "超级管理员", "系统初始化内置角色"),
        _upsert_role(db, "PURCHASER", "采购专员", "系统初始化内置角色"),
        _upsert_role(db, "WAREHOUSE", "仓库管理人员", "系统初始化内置角色"),
        _upsert_role(db, "SALES", "销售人员", "系统初始化内置角色"),
    ]

    def walk(nodes: list[tuple[str, str, list]]) -> list[tuple[str, str]]:
        result: list[tuple[str, str]] = []
        for code, name, children in nodes:
            result.append((code, name))
            result.extend(walk(children))
        return result

    permissions = []
    for code, name in walk(MENU_PERMISSION_TREE):
        row = db.scalar(select(Permission).where(Permission.permission_code == code))
        if row is None:
            row = Permission(
                permission_code=code,
                permission_name=name,
                module_code=code.replace("MENU_", ""),
                action_code="VIEW",
                status="ACTIVE",
                created_at=now,
                updated_at=now,
            )
            db.add(row)
            db.flush()
        else:
            row.permission_name = name
            row.status = "ACTIVE"
            row.updated_at = now
            db.add(row)
            db.flush()
        permissions.append(row)

    admin_role = roles[0]
    for permission in permissions:
        exists_row = db.scalar(
            select(RolePermission).where(
                RolePermission.role_id == admin_role.id,
                RolePermission.permission_id == permission.id,
            )
        )
        if exists_row is None:
            db.add(RolePermission(role_id=admin_role.id, permission_id=permission.id, created_at=now))
    return {"role_count": len(roles), "permission_count": len(permissions)}


def _seed_units_and_categories(db: Session) -> dict[str, int]:
    now = _now()
    unit_rows = [
        ("KG", "kg", 3),
        ("PCS", "件", 0),
        ("SET", "套", 0),
    ]
    for code, name, precision in unit_rows:
        row = db.execute(text("SELECT id FROM md_unit WHERE unit_code = :code"), {"code": code}).first()
        if row is None:
            db.execute(
                text(
                    """
                    INSERT INTO md_unit (unit_code, unit_name, precision_digits, created_at, updated_at)
                    VALUES (:code, :name, :precision, :now, :now)
                    """
                ),
                {"code": code, "name": name, "precision": precision, "now": now},
            )
    category_rows = [
        ("RAW", "原材料", "RAW_MATERIAL"),
        ("SEMI", "半成品", "SEMI_FINISHED"),
        ("FINISHED", "成品", "FINISHED_PRODUCT"),
        ("AUX", "辅料", "AUXILIARY"),
        ("SCRAP", "废料", "SCRAP"),
    ]
    for code, name, item_type in category_rows:
        row = db.execute(text("SELECT id FROM md_item_category WHERE category_code = :code"), {"code": code}).first()
        if row is None:
            db.execute(
                text(
                    """
                    INSERT INTO md_item_category
                      (category_code, category_name, parent_id, item_type, sort_no, status, created_at, updated_at)
                    VALUES (:code, :name, NULL, :item_type, 0, 'ACTIVE', :now, :now)
                    """
                ),
                {"code": code, "name": name, "item_type": item_type, "now": now},
            )
    return {"unit_count": len(unit_rows), "category_count": len(category_rows)}


def _seed_warehouses(db: Session) -> dict[str, int]:
    now = _now()
    for warehouse_type, warehouse_name, location_name in WAREHOUSE_TYPES:
        row = db.execute(
            text("SELECT id FROM wh_warehouse WHERE warehouse_type = :warehouse_type"),
            {"warehouse_type": warehouse_type},
        ).first()
        if row is None:
            result = db.execute(
                text(
                    """
                    INSERT INTO wh_warehouse
                      (warehouse_code, warehouse_name, warehouse_type, manager_employee_id, status, remark, created_at, updated_at)
                    VALUES (:code, :name, :warehouse_type, NULL, 'ACTIVE', '系统初始化默认仓库', :now, :now)
                    """
                ),
                {
                    "code": f"WH-{warehouse_type}",
                    "name": warehouse_name,
                    "warehouse_type": warehouse_type,
                    "now": now,
                },
            )
            warehouse_id = result.lastrowid
        else:
            warehouse_id = int(row[0])
            db.execute(
                text(
                    """
                    UPDATE wh_warehouse
                    SET warehouse_name = :name, status = 'ACTIVE', updated_at = :now
                    WHERE id = :warehouse_id
                    """
                ),
                {"name": warehouse_name, "warehouse_id": warehouse_id, "now": now},
            )
        location = db.execute(
            text("SELECT id FROM wh_location WHERE warehouse_id = :warehouse_id AND is_default = 1"),
            {"warehouse_id": warehouse_id},
        ).first()
        if location is None:
            db.execute(
                text(
                    """
                    INSERT INTO wh_location
                      (warehouse_id, location_code, location_name, zone_name, is_default, is_locked, status, remark, created_at, updated_at)
                    VALUES (:warehouse_id, :code, :name, NULL, 1, 0, 'ACTIVE', '系统初始化默认库位', :now, :now)
                    """
                ),
                {"warehouse_id": warehouse_id, "code": f"LOC-{warehouse_type}", "name": location_name, "now": now},
            )
    return {"warehouse_count": len(WAREHOUSE_TYPES), "location_count": len(WAREHOUSE_TYPES)}


def _seed_admin_user(db: Session, options: SystemInitializeOptions) -> dict[str, int | str]:
    from app.models.org import Department, Employee, Role, User, UserRole

    now = _now()
    root = db.scalar(select(Department).where(Department.dept_code == "ORG_ROOT"))
    if root is None:
        root = Department(
            dept_code="ORG_ROOT",
            dept_name=options.company_name,
            parent_id=None,
            org_node_type="COMPANY",
            dept_type="ADMIN",
            manager_name=None,
            manager_employee_id=None,
            status="ACTIVE",
            sort_no=0,
            remark="系统初始化根组织",
            created_at=now,
            updated_at=now,
        )
        db.add(root)
        db.flush()
    else:
        root.dept_name = options.company_name
        root.org_node_type = "COMPANY"
        root.dept_type = "ADMIN"
        root.status = "ACTIVE"
        root.updated_at = now
        db.add(root)
        db.flush()

    employee = db.scalar(select(Employee).where(Employee.employee_code == "EMP-ADMIN-001"))
    if employee is None:
        employee = Employee(
            employee_code="EMP-ADMIN-001",
            employee_name=options.admin_name,
            dept_id=root.id,
            mobile=options.admin_phone,
            gender=None,
            hire_date=None,
            job_title="超级管理员",
            shift_code=None,
            manager_employee_id=None,
            is_operator=0,
            is_workshop_staff=0,
            status="ACTIVE",
            remark="系统初始化超级管理员",
            created_at=now,
            updated_at=now,
        )
        db.add(employee)
        db.flush()
    else:
        employee.employee_name = options.admin_name
        employee.dept_id = root.id
        employee.mobile = options.admin_phone
        employee.status = "ACTIVE"
        employee.updated_at = now
        db.add(employee)
        db.flush()

    user = db.scalar(select(User).where(User.username == options.admin_phone))
    if user is None:
        user = User(
            username=options.admin_phone,
            password_hash=hash_password(options.admin_password),
            employee_id=employee.id,
            dept_id=root.id,
            nickname=options.admin_name,
            email=None,
            is_super_admin=1,
            last_login_at=None,
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        db.add(user)
        db.flush()
    else:
        user.password_hash = hash_password(options.admin_password)
        user.employee_id = employee.id
        user.dept_id = root.id
        user.nickname = options.admin_name
        user.is_super_admin = 1
        user.status = "ACTIVE"
        user.updated_at = now
        db.add(user)
        db.flush()

    admin_role = db.scalar(select(Role).where(Role.role_code == "ADMIN"))
    if admin_role is not None:
        existing = db.scalar(select(UserRole).where(UserRole.user_id == user.id, UserRole.role_id == admin_role.id))
        if existing is None:
            db.add(UserRole(user_id=user.id, role_id=admin_role.id, created_at=now))
    return {"admin_user_id": user.id, "admin_username": user.username}


def _seed_miniapp_defaults(db: Session, options: SystemInitializeOptions) -> dict[str, int]:
    now = _now()
    attendance_point = db.execute(
        text("SELECT name FROM attendance_points WHERE name = :name"),
        {"name": options.company_name},
    ).first()
    if attendance_point is None:
        db.execute(
            text(
                """
                INSERT INTO attendance_points
                  (name, latitude, longitude, radius_meters, remark, is_active, created_at, updated_at)
                VALUES (:name, NULL, NULL, 500, '系统初始化默认考勤点', 1, :now, :now)
                """
            ),
            {"name": options.company_name, "now": now},
        )
    else:
        db.execute(
            text(
                """
                UPDATE attendance_points
                SET remark = '系统初始化默认考勤点', is_active = 1, updated_at = :now
                WHERE name = :name
                """
            ),
            {"name": options.company_name, "now": now},
        )

    personnel = db.execute(
        text("SELECT phone FROM personnel WHERE phone = :phone"),
        {"phone": options.admin_phone},
    ).first()
    if personnel is None:
        db.execute(
            text(
                """
                INSERT INTO personnel (phone, name, is_temporary, temporary_expires_at, created_at, updated_at)
                VALUES (:phone, :name, 0, NULL, :now, :now)
                """
            ),
            {"phone": options.admin_phone, "name": options.admin_name, "now": now},
        )
    else:
        db.execute(
            text(
                """
                UPDATE personnel
                SET name = :name, is_temporary = 0, temporary_expires_at = NULL, updated_at = :now
                WHERE phone = :phone
                """
            ),
            {"phone": options.admin_phone, "name": options.admin_name, "now": now},
        )

    role = db.execute(
        text("SELECT phone FROM person_roles WHERE phone = :phone AND role = 'admin'"),
        {"phone": options.admin_phone},
    ).first()
    if role is None:
        db.execute(
            text("INSERT INTO person_roles (phone, role, created_at) VALUES (:phone, 'admin', :now)"),
            {"phone": options.admin_phone, "now": now},
        )

    binding = db.execute(
        text(
            """
            SELECT phone
            FROM person_attendance_points
            WHERE phone = :phone AND attendance_point_name = :point
            """
        ),
        {"phone": options.admin_phone, "point": options.company_name},
    ).first()
    if binding is None:
        db.execute(
            text(
                """
                INSERT INTO person_attendance_points (phone, attendance_point_name, created_at)
                VALUES (:phone, :point, :now)
                """
            ),
            {"phone": options.admin_phone, "point": options.company_name, "now": now},
        )
    return {"miniapp_attendance_point_count": 1, "miniapp_admin_count": 1}


def seed_system_skeleton(db: Session, options: SystemInitializeOptions) -> dict[str, int | str | bool]:
    permission_summary = _seed_permissions_and_roles(db)
    config_enabled = _bool_to_config_value(options.smart_operation_report_enabled)
    _upsert_system_config(
        db,
        code="RAW_MATERIAL_LOT_PREFIX",
        name="原材料库存批次前缀",
        value="YL",
        remark="系统初始化默认值，可在系统管理中维护",
    )
    _upsert_system_config(
        db,
        code="SMART_OPERATION_REPORT_ENABLED",
        name="对接智能报工小程序",
        value=config_enabled,
        remark="开启后显示工序报工并同步小程序报工数据",
    )
    unit_summary = _seed_units_and_categories(db)
    warehouse_summary = _seed_warehouses(db)
    admin_summary = _seed_admin_user(db, options)
    miniapp_summary = _seed_miniapp_defaults(db, options)
    return {
        **permission_summary,
        **unit_summary,
        **warehouse_summary,
        **admin_summary,
        **miniapp_summary,
        "smart_operation_report_enabled": options.smart_operation_report_enabled,
    }

This implementation intentionally uses SELECT + INSERT / UPDATE instead of MySQL-only ON DUPLICATE KEY UPDATE, so the same seed tests pass on SQLite and MySQL.

• Step 4: Run skeleton seed test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_seed.py -q

Expected: PASS.

• Step 5: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/services/system_initializer.py backend/tests/test_system_initializer_seed.py
git commit -m "feat: seed system initialization skeleton"

---

Task 4: Implement Reset Safety, Counting, And Truncation

Files:

• Modify: backend/app/services/system_initializer.py

• Create: backend/tests/test_system_initializer_reset.py

• Step 1: Write failing reset protection test

Create backend/tests/test_system_initializer_reset.py:

from __future__ import annotations

import unittest

from sqlalchemy import BigInteger, create_engine, select, text
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import Session, sessionmaker


@compiles(BigInteger, "sqlite")
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
    _ = type_, compiler, kw
    return "INTEGER"


import app.models.document_archive  # noqa: E402,F401
import app.models.master_data  # noqa: E402,F401
import app.models.miniapp  # noqa: E402,F401
import app.models.operations  # noqa: E402,F401
import app.models.org  # noqa: E402,F401
import app.models.planning  # noqa: E402,F401
import app.models.sales  # noqa: E402,F401
from app.models.base import Base  # noqa: E402
from app.models.org import Department, User  # noqa: E402


class SystemInitializerResetTest(unittest.TestCase):
    def setUp(self) -> None:
        engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
        Base.metadata.create_all(engine)
        self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
        self.db: Session = self.SessionLocal()

    def tearDown(self) -> None:
        self.db.close()

    def test_reset_requires_explicit_confirmation(self) -> None:
        from app.services.system_initializer import SystemInitializeOptions, reset_existing_database

        options = SystemInitializeOptions(
            mode="reset",
            company_name="百华",
            admin_name="超级管理员",
            admin_phone="13800000000",
            admin_password="secret123",
            confirm_reset=False,
        )
        with self.assertRaisesRegex(ValueError, "confirm-reset"):
            reset_existing_database(self.db, options)

    def test_reset_clears_test_rows_and_rebuilds_admin(self) -> None:
        from app.services.system_initializer import SystemInitializeOptions, reset_existing_database

        self.db.execute(
            text(
                """
            INSERT INTO md_customer
              (customer_code, customer_name, status, credit_days, created_at, updated_at)
            VALUES ('CUS-TEST', '嘉恒测试客户', 'ACTIVE', 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """
            )
        )
        self.db.execute(
            text(
                """
            INSERT INTO sys_department
              (dept_code, dept_name, org_node_type, dept_type, status, sort_no, created_at, updated_at)
            VALUES ('OLD', '嘉恒测试部门', 'DEPARTMENT', 'ADMIN', 'ACTIVE', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """
            )
        )
        self.db.commit()

        options = SystemInitializeOptions(
            mode="reset",
            company_name="百华",
            admin_name="超级管理员",
            admin_phone="13800000000",
            admin_password="secret123",
            smart_operation_report_enabled=True,
            confirm_reset=True,
        )
        result = reset_existing_database(self.db, options)
        self.db.commit()

        self.assertEqual(self.db.execute(text("SELECT COUNT(*) FROM md_customer")).scalar(), 0)
        self.assertIsNone(self.db.scalar(select(Department).where(Department.dept_code == "OLD")))
        admin_user = self.db.scalar(select(User).where(User.username == "13800000000"))
        self.assertIsNotNone(admin_user)
        self.assertEqual(admin_user.is_super_admin, 1)
        self.assertEqual(result.seeded["admin_username"], "13800000000")

• Step 2: Run reset tests to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_reset.py -q

Expected: FAIL because reset_existing_database does not exist.

• Step 3: Implement table helpers and reset flow

Append to backend/app/services/system_initializer.py:

def _existing_tables(db: Session) -> set[str]:
    return set(inspect(db.bind).get_table_names())


def _count_table(db: Session, table_name: str) -> int:
    return int(db.execute(text(f"SELECT COUNT(*) FROM `{table_name}`")).scalar() or 0)


def count_existing_tables(db: Session, table_names: Iterable[str]) -> dict[str, int]:
    existing = _existing_tables(db)
    return {table: _count_table(db, table) for table in table_names if table in existing}


def _delete_all_rows(db: Session, table_name: str) -> None:
    db.execute(text(f"DELETE FROM `{table_name}`"))


def _reset_autoincrement_if_supported(db: Session, table_name: str) -> None:
    dialect = db.bind.dialect.name if db.bind is not None else ""
    if dialect == "mysql":
        db.execute(text(f"ALTER TABLE `{table_name}` AUTO_INCREMENT = 1"))
    elif dialect == "sqlite":
        db.execute(text("DELETE FROM sqlite_sequence WHERE name = :table_name"), {"table_name": table_name})


def _clear_tables(db: Session, table_names: Iterable[str]) -> None:
    existing = _existing_tables(db)
    dialect = db.bind.dialect.name if db.bind is not None else ""
    if dialect == "mysql":
        db.execute(text("SET FOREIGN_KEY_CHECKS=0"))
    try:
        for table_name in table_names:
            if table_name not in existing:
                continue
            _delete_all_rows(db, table_name)
            _reset_autoincrement_if_supported(db, table_name)
    finally:
        if dialect == "mysql":
            db.execute(text("SET FOREIGN_KEY_CHECKS=1"))


def reset_existing_database(db: Session, options: SystemInitializeOptions) -> SystemInitializeResult:
    if not options.confirm_reset:
        raise ValueError("reset mode requires --confirm-reset")
    tables_to_clear = BUSINESS_RESET_TABLES + MINIAPP_RESET_TABLES + ACCOUNT_RESET_TABLES
    counts_before = count_existing_tables(db, tables_to_clear + SKELETON_TABLES)
    _clear_tables(db, tables_to_clear)
    seeded = seed_system_skeleton(db, options)
    counts_after = count_existing_tables(db, tables_to_clear + SKELETON_TABLES)
    return SystemInitializeResult(
        mode="reset",
        database="",
        backup_path=None,
        summary_path=None,
        counts_before=counts_before,
        counts_after=counts_after,
        seeded=seeded,
    )

If SQLite raises on raw SQL strings in the test, wrap raw inserts with sqlalchemy.text(...).

• Step 4: Run reset tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_reset.py tests/test_system_initializer_seed.py -q

Expected: PASS.

• Step 5: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/services/system_initializer.py backend/tests/test_system_initializer_reset.py
git commit -m "feat: reset existing system data safely"

---

Task 5: Implement Fresh Schema Creation

Files:

• Modify: backend/app/services/system_initializer.py

• Modify: backend/tests/test_system_initializer_seed.py

• Step 1: Add failing fresh create-all test

Append to backend/tests/test_system_initializer_seed.py:

class SystemInitializerFreshTest(unittest.TestCase):
    def test_create_all_tables_on_engine_registers_required_tables(self) -> None:
        from app.services.system_initializer import create_all_schema_tables

        engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
        create_all_schema_tables(engine)
        inspector = inspect(engine)
        table_names = set(inspector.get_table_names())
        self.assertIn("sys_user", table_names)
        self.assertIn("wh_stock_lot", table_names)
        self.assertIn("work_sessions", table_names)
        self.assertIn("report_audit_logs", table_names)

Also import inspect at the top:

from sqlalchemy import BigInteger, create_engine, inspect, select

• Step 2: Run fresh test to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_seed.py::SystemInitializerFreshTest::test_create_all_tables_on_engine_registers_required_tables -q

Expected: FAIL because create_all_schema_tables does not exist.

• Step 3: Implement model import and create-all helpers

Append to backend/app/services/system_initializer.py:

def import_all_models() -> None:
    import app.models.document_archive  # noqa: F401
    import app.models.master_data  # noqa: F401
    import app.models.miniapp  # noqa: F401
    import app.models.operations  # noqa: F401
    import app.models.org  # noqa: F401
    import app.models.planning  # noqa: F401
    import app.models.sales  # noqa: F401


def create_all_schema_tables(engine: Engine) -> None:
    import_all_models()
    Base.metadata.create_all(engine)

• Step 4: Implement MySQL database creation helper

Append:

def create_database_if_missing(settings: Settings) -> None:
    server_url = settings.mysql_dsn.set(database=None)
    server_engine = create_engine(server_url, pool_pre_ping=True, future=True)
    with server_engine.begin() as conn:
        conn.execute(
            text(
                f"CREATE DATABASE IF NOT EXISTS `{settings.mysql_database}` "
                "DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_unicode_ci"
            )
        )
    server_engine.dispose()


def fresh_initialize_database(engine: Engine, options: SystemInitializeOptions) -> SystemInitializeResult:
    create_all_schema_tables(engine)
    with Session(engine) as db:
        seeded = seed_system_skeleton(db, options)
        db.commit()
        counts_after = count_existing_tables(db, SKELETON_TABLES + MINIAPP_RESET_TABLES)
    return SystemInitializeResult(
        mode="fresh",
        database="",
        backup_path=None,
        summary_path=None,
        counts_before={},
        counts_after=counts_after,
        seeded=seeded,
    )

• Step 5: Run fresh and seed tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_models.py tests/test_system_initializer_seed.py -q

Expected: PASS.

• Step 6: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/services/system_initializer.py backend/tests/test_system_initializer_seed.py
git commit -m "feat: create fresh system schema"

---

Task 6: Add Backup And Summary Report Generation

Files:

• Modify: backend/app/services/system_initializer.py

• Modify: backend/tests/test_system_initializer_reset.py

• Step 1: Add failing summary write test

Append to backend/tests/test_system_initializer_reset.py:

import tempfile
from pathlib import Path


class SystemInitializerReportTest(unittest.TestCase):
    def test_write_summary_file_contains_mode_and_counts(self) -> None:
        from app.services.system_initializer import SystemInitializeResult, write_summary_report

        with tempfile.TemporaryDirectory() as tmpdir:
            result = SystemInitializeResult(
                mode="reset",
                database="jiaheng_erp",
                backup_path="/tmp/backup.sql",
                summary_path=None,
                counts_before={"md_customer": 2},
                counts_after={"md_customer": 0},
                seeded={"admin_username": "13800000000"},
            )
            output = write_summary_report(Path(tmpdir), result)
            content = output.read_text(encoding="utf-8")
            self.assertIn('"mode": "reset"', content)
            self.assertIn('"md_customer": 2', content)
            self.assertIn('"admin_username": "13800000000"', content)

• Step 2: Run report test to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_reset.py::SystemInitializerReportTest::test_write_summary_file_contains_mode_and_counts -q

Expected: FAIL because write_summary_report does not exist.

• Step 3: Implement report writing

Append to backend/app/services/system_initializer.py:

def _timestamp() -> str:
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def default_output_dir() -> Path:
    return Path(__file__).resolve().parents[3] / "outputs" / "cleanup_backups"


def write_summary_report(output_dir: Path, result: SystemInitializeResult) -> Path:
    output_dir.mkdir(parents=True, exist_ok=True)
    summary_path = output_dir / f"system_init_summary_{_timestamp()}.json"
    payload = {
        "mode": result.mode,
        "database": result.database,
        "backup_path": result.backup_path,
        "summary_path": str(summary_path),
        "counts_before": result.counts_before,
        "counts_after": result.counts_after,
        "seeded": result.seeded,
        "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }
    summary_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    return summary_path

• Step 4: Implement MySQL backup helper

Append:

def backup_mysql_database(settings: Settings, output_dir: Path) -> Path:
    output_dir.mkdir(parents=True, exist_ok=True)
    backup_path = output_dir / f"system_init_backup_{_timestamp()}.sql"
    command = [
        "mysqldump",
        "--default-character-set=utf8mb4",
        "-h",
        settings.mysql_host,
        "-P",
        str(settings.mysql_port),
        "-u",
        settings.mysql_user,
        f"-p{settings.mysql_password}",
        settings.mysql_database,
    ]
    with backup_path.open("wb") as handle:
        subprocess.run(command, stdout=handle, stderr=subprocess.PIPE, check=True)
    return backup_path

This helper is only used for MySQL-backed CLI execution. Unit tests do not run mysqldump.

• Step 5: Wire reset result to include backup/report in service helper

Append:

def reset_existing_database_with_backup(
    engine: Engine,
    settings: Settings,
    options: SystemInitializeOptions,
    *,
    output_dir: Path | None = None,
) -> SystemInitializeResult:
    backup_dir = output_dir or default_output_dir()
    backup_path = backup_mysql_database(settings, backup_dir)
    with Session(engine) as db:
        result = reset_existing_database(db, options)
        db.commit()
    result = SystemInitializeResult(
        mode=result.mode,
        database=settings.mysql_database,
        backup_path=str(backup_path),
        summary_path=None,
        counts_before=result.counts_before,
        counts_after=result.counts_after,
        seeded=result.seeded,
    )
    summary_path = write_summary_report(backup_dir, result)
    return SystemInitializeResult(
        mode=result.mode,
        database=result.database,
        backup_path=result.backup_path,
        summary_path=str(summary_path),
        counts_before=result.counts_before,
        counts_after=result.counts_after,
        seeded=result.seeded,
    )

• Step 6: Run report and reset tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_reset.py -q

Expected: PASS.

• Step 7: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/services/system_initializer.py backend/tests/test_system_initializer_reset.py
git commit -m "feat: add initialization backup and reports"

---

Task 7: Add CLI Entry Point

Files:

• Create: backend/scripts/system_initialize.py

• Create: backend/tests/test_system_initializer_cli.py

• Step 1: Write failing CLI parse test

Create backend/tests/test_system_initializer_cli.py:

from __future__ import annotations

import unittest


class SystemInitializerCliTest(unittest.TestCase):
    def test_parse_reset_arguments(self) -> None:
        from scripts.system_initialize import parse_args

        args = parse_args(
            [
                "reset",
                "--company-name",
                "百华",
                "--admin-name",
                "超级管理员",
                "--admin-phone",
                "13800000000",
                "--admin-password",
                "secret123",
                "--confirm-reset",
                "--smart-operation-report",
                "off",
            ]
        )

        self.assertEqual(args.mode, "reset")
        self.assertEqual(args.company_name, "百华")
        self.assertEqual(args.admin_phone, "13800000000")
        self.assertTrue(args.confirm_reset)
        self.assertEqual(args.smart_operation_report, "off")


if __name__ == "__main__":
    unittest.main()

• Step 2: Run CLI test to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_cli.py -q

Expected: FAIL because backend/scripts/system_initialize.py does not exist.

• Step 3: Create scripts package and CLI

Create directory backend/scripts if it does not exist.

Create backend/scripts/__init__.py:

"""Operational scripts for ForgeFlow ERP backend."""

Create backend/scripts/system_initialize.py:

from __future__ import annotations

import argparse
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from sqlalchemy import create_engine  # noqa: E402

from app.core.config import get_settings  # noqa: E402
from app.services.system_initializer import (  # noqa: E402
    SystemInitializeOptions,
    create_database_if_missing,
    fresh_initialize_database,
    reset_existing_database_with_backup,
)


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Initialize ForgeFlow ERP database for customer deployment.")
    subparsers = parser.add_subparsers(dest="mode", required=True)

    def add_common(subparser: argparse.ArgumentParser) -> None:
        subparser.add_argument("--company-name", required=True)
        subparser.add_argument("--admin-name", required=True)
        subparser.add_argument("--admin-phone", required=True)
        subparser.add_argument("--admin-password", required=True)
        subparser.add_argument(
            "--smart-operation-report",
            choices=["on", "off"],
            default="on",
            help="Whether to enable smart miniapp operation report integration.",
        )
        subparser.add_argument("--delete-files", action="store_true")
        subparser.add_argument("--allow-any-database", action="store_true")

    fresh = subparsers.add_parser("fresh", help="Create a new database/schema and seed clean system skeleton.")
    add_common(fresh)

    reset = subparsers.add_parser("reset", help="Reset an existing database after backup.")
    add_common(reset)
    reset.add_argument("--confirm-reset", action="store_true")

    return parser.parse_args(argv)


def _options_from_args(args: argparse.Namespace) -> SystemInitializeOptions:
    return SystemInitializeOptions(
        mode=args.mode,
        company_name=args.company_name,
        admin_name=args.admin_name,
        admin_phone=args.admin_phone,
        admin_password=args.admin_password,
        smart_operation_report_enabled=args.smart_operation_report == "on",
        confirm_reset=bool(getattr(args, "confirm_reset", False)),
        delete_files=bool(args.delete_files),
        allow_any_database=bool(args.allow_any_database),
    )


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    settings = get_settings()
    options = _options_from_args(args)
    print(f"mode={options.mode}")
    print(f"database={settings.mysql_host}:{settings.mysql_port}/{settings.mysql_database}")

    if options.mode == "fresh":
        create_database_if_missing(settings)
        engine = create_engine(settings.mysql_dsn, pool_pre_ping=True, future=True)
        result = fresh_initialize_database(engine, options)
    else:
        engine = create_engine(settings.mysql_dsn, pool_pre_ping=True, future=True)
        result = reset_existing_database_with_backup(engine, settings, options)

    print(f"backup_path={result.backup_path or ''}")
    print(f"summary_path={result.summary_path or ''}")
    print(f"admin_username={result.seeded.get('admin_username', '')}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

• Step 4: Run CLI test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_cli.py -q

Expected: PASS.

• Step 5: Run CLI help manually

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python scripts/system_initialize.py --help
.venv/bin/python scripts/system_initialize.py reset --help

Expected: Both commands print usage text and exit 0.

• Step 6: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/scripts backend/tests/test_system_initializer_cli.py
git commit -m "feat: add system initialization CLI"

---

Task 8: Add Optional File Cleanup

Files:

• Modify: backend/app/services/system_initializer.py

• Modify: backend/tests/test_system_initializer_reset.py

• Step 1: Add file cleanup test

Append to backend/tests/test_system_initializer_reset.py:

class SystemInitializerFileCleanupTest(unittest.TestCase):
    def test_delete_managed_files_removes_only_known_children(self) -> None:
        from app.services.system_initializer import delete_managed_files

        with tempfile.TemporaryDirectory() as tmpdir:
            root = Path(tmpdir)
            archive_dir = root / "document_archives"
            photo_dir = root / "uploads" / "logistics"
            archive_dir.mkdir(parents=True)
            photo_dir.mkdir(parents=True)
            (archive_dir / "a.pdf").write_text("pdf", encoding="utf-8")
            (photo_dir / "b.png").write_text("png", encoding="utf-8")

            summary = delete_managed_files([archive_dir, photo_dir])

            self.assertEqual(summary["deleted_file_count"], 2)
            self.assertFalse((archive_dir / "a.pdf").exists())
            self.assertFalse((photo_dir / "b.png").exists())
            self.assertTrue(archive_dir.exists())
            self.assertTrue(photo_dir.exists())

• Step 2: Run file cleanup test to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_reset.py::SystemInitializerFileCleanupTest::test_delete_managed_files_removes_only_known_children -q

Expected: FAIL because delete_managed_files does not exist.

• Step 3: Implement file cleanup helper

Append to backend/app/services/system_initializer.py:

def delete_managed_files(directories: list[Path]) -> dict[str, int]:
    deleted_file_count = 0
    deleted_dir_count = 0
    for directory in directories:
        if not directory.exists() or not directory.is_dir():
            continue
        for child in directory.iterdir():
            if child.is_file():
                child.unlink()
                deleted_file_count += 1
            elif child.is_dir():
                shutil.rmtree(child)
                deleted_dir_count += 1
    return {"deleted_file_count": deleted_file_count, "deleted_dir_count": deleted_dir_count}

Do not call this helper unless options.delete_files is True.

• Step 4: Wire file cleanup into CLI result

Modify reset_existing_database_with_backup after write_summary_report:

    file_cleanup = {}
    if options.delete_files:
        root = Path(__file__).resolve().parents[3]
        file_cleanup = delete_managed_files(
            [
                root / "outputs" / "document_archives",
                root / "outputs" / "document_archive_batches",
                root / "backend" / "uploads" / "logistics",
            ]
        )
        result.seeded.update(file_cleanup)

If SystemInitializeResult.seeded is frozen through dataclass immutability concerns, construct a new seeded = {**result.seeded, **file_cleanup} dict and pass it into the returned result.

• Step 5: Run file cleanup tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_system_initializer_reset.py -q

Expected: PASS.

• Step 6: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/services/system_initializer.py backend/tests/test_system_initializer_reset.py
git commit -m "feat: support optional initialization file cleanup"

---

Task 9: Add End-To-End Local Verification Commands

Files:

• Create: docs/superpowers/plans/2026-06-14-system-initialization-bootstrap-runbook.md

• Modify: docs/superpowers/plans/2026-06-14-system-initialization-bootstrap.md

• Step 1: Create runbook

Create docs/superpowers/plans/2026-06-14-system-initialization-bootstrap-runbook.md:

# System Initialization Runbook

## Fresh Initialization On A New Database

```bash
cd /home/souplearn/ERP/ForgeFlow-ERP/backend
source .venv/bin/activate

python scripts/system_initialize.py fresh \
  --company-name 百华 \
  --admin-name 超级管理员 \
  --admin-phone 13800000000 \
  --admin-password '正式密码' \
  --smart-operation-report on

Reset Existing Test Database

cd /home/souplearn/ERP/ForgeFlow-ERP/backend
source .venv/bin/activate

python scripts/system_initialize.py reset \
  --company-name 百华 \
  --admin-name 超级管理员 \
  --admin-phone 13800000000 \
  --admin-password '正式密码' \
  --smart-operation-report on \
  --confirm-reset

Verification

curl -s http://127.0.0.1:8000/api/system/health | python3 -m json.tool

mysql --default-character-set=utf8mb4 -h "$MYSQL_HOST" -P "$MYSQL_PORT" -u "$MYSQL_USER" -p"$MYSQL_PASSWORD" "$MYSQL_DATABASE" -e "
SELECT COUNT(*) AS customer_count FROM md_customer;
SELECT COUNT(*) AS supplier_count FROM md_supplier;
SELECT COUNT(*) AS stock_lot_count FROM wh_stock_lot;
SELECT COUNT(*) AS sales_order_count FROM so_sales_order;
SELECT COUNT(*) AS admin_count FROM sys_user WHERE username='13800000000' AND is_super_admin=1;
"

Expected:

• Health endpoint reports database connected.
• Business tables return 0.
• Admin count returns 1.
• ERP login succeeds with the initialized admin phone and password.

- [ ] **Step 2: Add runbook pointer to implementation plan**

Append to this implementation plan:

```markdown
## Runbook

Deployment commands are maintained in:

- `docs/superpowers/plans/2026-06-14-system-initialization-bootstrap-runbook.md`

• Step 3: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add docs/superpowers/plans/2026-06-14-system-initialization-bootstrap-runbook.md docs/superpowers/plans/2026-06-14-system-initialization-bootstrap.md
git commit -m "docs: add system initialization runbook"

---

Task 10: Full Verification Before Server Use

Files:

• No source changes expected.

• Step 1: Run focused tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest \
  tests/test_system_initializer_models.py \
  tests/test_system_initializer_seed.py \
  tests/test_system_initializer_reset.py \
  tests/test_system_initializer_cli.py \
  -q

Expected: PASS.

• Step 2: Run broader backend smoke tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest \
  tests/test_system_permission_management.py \
  tests/test_smart_operation_report_config.py \
  tests/test_selected_stock_lot_production_issue.py \
  tests/test_purchase_order_document_archive.py \
  tests/test_warehouse_operation_document_archive.py \
  -q

Expected: PASS.

• Step 3: Run frontend build

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build

Expected: PASS.

• Step 4: Review git diff

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short --branch
git log --oneline -n 8

Expected: working tree clean after task commits, branch on BH_DEV.

• Step 5: Ask user before running against server database

Do not execute reset against the server database without explicit user approval at action time. The next message must state the exact command, target host/database, whether files will be deleted, and backup path behavior.

---

Spec Coverage Self-Review

• fresh full database/schema initialization is covered by Tasks 1, 3, 5, 7, and 10.
• reset existing database initialization is covered by Tasks 2, 3, 4, 6, 7, 8, and 10.
• ERP table coverage is covered by Task 1 metadata tests and Task 5 create_all_schema_tables.
• Miniapp table coverage is covered by Task 1 model expansion and tests.
• "Only keep one super admin" is covered by Task 3 seed and Task 4 reset test.
• Backup and JSON report are covered by Task 6.
• Optional file deletion is covered by Task 8 and defaults to off.
• Runbook for future customer deployment is covered by Task 9.
• Server execution safety gate is covered by Task 10.

Runbook

Deployment commands are maintained in:

• docs/superpowers/plans/2026-06-14-system-initialization-bootstrap-runbook.md