107 lines
5.5 KiB
Python
107 lines
5.5 KiB
Python
"""Prepare a fresh MySQL database for the advisor demonstration.
|
|
|
|
This is deliberately an orchestrator, not a database dump tool. The legacy
|
|
``jr_agent`` database has schema drift and must not be copied or upgraded in
|
|
place. The command is dry-run by default and requires two explicit flags
|
|
before it writes the configured target database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Step:
|
|
label: str
|
|
command: tuple[str, ...]
|
|
display: str
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Bootstrap an isolated advisor demo database")
|
|
parser.add_argument("--apply", action="store_true", help="Run commands that write the target database")
|
|
parser.add_argument(
|
|
"--confirm-demo-database",
|
|
action="store_true",
|
|
help="Confirm that MYSQL_DSN points to a disposable demo database",
|
|
)
|
|
parser.add_argument("--suitability-csv", type=Path, help="Reviewed suitability disclosure CSV")
|
|
parser.add_argument("--contracts-csv", type=Path, help="Reviewed fund-contract disclosure CSV")
|
|
parser.add_argument("--market-days", type=int, default=400, help="Days of history to synchronize")
|
|
parser.add_argument("--skip-market-sync", action="store_true", help="Do not call public market sources")
|
|
return parser.parse_args()
|
|
|
|
|
|
def build_steps(args: argparse.Namespace) -> tuple[Step, ...]:
|
|
python = sys.executable
|
|
steps = [
|
|
Step("Apply current Alembic migrations", (python, "-m", "alembic", "upgrade", "head"), "python -m alembic upgrade head"),
|
|
Step("Seed baseline demo RBAC", (python, "tools/seed_test_rbac.py"), "python tools/seed_test_rbac.py"),
|
|
Step("Set customer login", (python, "tools/create_test_user.py", "--id", "9001", "--username", "cust_t", "--role", "customer", "--password", "123456"), "python tools/create_test_user.py --id 9001 --username cust_t --role customer --password <demo-password>"),
|
|
Step("Set administrator login", (python, "tools/create_test_user.py", "--id", "9003", "--username", "admin_t", "--role", "admin", "--password", "88888888"), "python tools/create_test_user.py --id 9003 --username admin_t --role admin --password <demo-password>"),
|
|
Step("Create advisor role and permissions", (python, "tools/grant_advisor_role.py"), "python tools/grant_advisor_role.py"),
|
|
Step("Set advisor login", (python, "tools/create_test_user.py", "--id", "9020", "--username", "advisor_t", "--role", "advisor", "--password", "abc12345"), "python tools/create_test_user.py --id 9020 --username advisor_t --role advisor --password <demo-password>"),
|
|
Step("Seed compliance baseline", (python, "tools/seed_compliance_baseline.py"), "python tools/seed_compliance_baseline.py"),
|
|
Step("Seed customer assessment and advisor assignment", (python, "tools/seed_advisor_demo.py"), "python tools/seed_advisor_demo.py"),
|
|
]
|
|
if args.suitability_csv is not None:
|
|
steps.append(
|
|
Step(
|
|
"Import reviewed product evidence",
|
|
(python, "tools/import_product_governance_reference.py", "--suitability", str(args.suitability_csv), "--contracts", str(args.contracts_csv)),
|
|
"python tools/import_product_governance_reference.py --suitability <reviewed.csv> --contracts <reviewed.csv>",
|
|
)
|
|
)
|
|
steps.append(
|
|
Step("Classify product assets", (python, "tools/import_product_asset_classifications.py"), "python tools/import_product_asset_classifications.py")
|
|
)
|
|
if not args.skip_market_sync:
|
|
steps.append(
|
|
Step("Refresh market history and quality snapshots", (python, "tools/sync_advisor_market_data.py", "--days", str(args.market_days)), f"python tools/sync_advisor_market_data.py --days {args.market_days}"),
|
|
)
|
|
steps.extend((
|
|
Step("Publish advisor tool allowlist", (python, "tools/publish_advisor_demo_config.py", "--apply"), "python tools/publish_advisor_demo_config.py --apply"),
|
|
Step("Audit database schema", (python, "tools/audit_schema.py"), "python tools/audit_schema.py"),
|
|
Step("Audit database constraints", (python, "tools/audit_constraints.py"), "python tools/audit_constraints.py"),
|
|
))
|
|
return tuple(steps)
|
|
|
|
|
|
def validate_args(args: argparse.Namespace) -> str | None:
|
|
if (args.suitability_csv is None) != (args.contracts_csv is None):
|
|
return "--suitability-csv and --contracts-csv must be supplied together"
|
|
if args.market_days < 20:
|
|
return "--market-days must be at least 20"
|
|
if args.apply and not args.confirm_demo_database:
|
|
return "--apply requires --confirm-demo-database"
|
|
return None
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
error = validate_args(args)
|
|
if error:
|
|
print(f"error: {error}")
|
|
return 2
|
|
steps = build_steps(args)
|
|
print("Target database must be newly created and disposable; never use legacy jr_agent.")
|
|
if args.suitability_csv is None:
|
|
print("warning: no reviewed product evidence supplied; recommendations remain fail-closed.")
|
|
for index, step in enumerate(steps, start=1):
|
|
print(f"[{index}/{len(steps)}] {step.label}\n $ {step.display}")
|
|
if args.apply and subprocess.run(step.command, check=False).returncode != 0:
|
|
print(f"failed: {step.label}")
|
|
return 1
|
|
if not args.apply:
|
|
print("dry run complete; use --apply --confirm-demo-database after checking MYSQL_DSN and .env.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|