90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
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_arguments(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="Enable or disable 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 database if needed, create schema, and seed a clean skeleton.")
|
|
add_common_arguments(fresh)
|
|
|
|
reset = subparsers.add_parser("reset", help="Back up and reset an existing database into a clean skeleton.")
|
|
add_common_arguments(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 _print_result(settings: object, result: object, options: SystemInitializeOptions) -> None:
|
|
print(f"mode={options.mode}")
|
|
print(f"database={settings.mysql_host}:{settings.mysql_port}/{settings.mysql_database}")
|
|
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', '')}")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
settings = get_settings()
|
|
options = _options_from_args(args)
|
|
engine = create_engine(settings.mysql_dsn, pool_pre_ping=True, future=True)
|
|
|
|
if options.mode == "fresh":
|
|
create_database_if_missing(settings)
|
|
result = fresh_initialize_database(engine, options)
|
|
else:
|
|
result = reset_existing_database_with_backup(engine, settings, options)
|
|
|
|
_print_result(settings, result, options)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|