75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
"""开发与演示命令行。"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
|
|
from sqlalchemy import select
|
|
|
|
from .db import drop_db, init_db, session_scope
|
|
from .models import EvidenceBundle, Run
|
|
from .schemas import CreateRunBody
|
|
from .seed import ensure_seed_data
|
|
from .services.runs import create_run, run_to_dict, simulate_run
|
|
|
|
|
|
async def _init_db() -> None:
|
|
await init_db()
|
|
|
|
|
|
async def _seed() -> None:
|
|
await init_db()
|
|
async with session_scope() as session:
|
|
await ensure_seed_data(session)
|
|
|
|
|
|
async def _reset() -> None:
|
|
await drop_db()
|
|
await init_db()
|
|
await _seed()
|
|
|
|
|
|
async def _demo() -> None:
|
|
await _seed()
|
|
async with session_scope() as session:
|
|
run = await create_run(
|
|
session,
|
|
CreateRunBody(loops=8, inject_failure=True, created_by="cli-demo"),
|
|
)
|
|
run_id = run.id
|
|
print("Run {0} 已启动：将注入一次心跳丢失并自动升级到 L3 恢复。".format(run_id))
|
|
await simulate_run(run_id)
|
|
async with session_scope() as session:
|
|
run = await session.get(Run, run_id)
|
|
bundle = await session.scalar(
|
|
select(EvidenceBundle)
|
|
.where(EvidenceBundle.run_id == run_id)
|
|
.order_by(EvidenceBundle.created_at.desc())
|
|
)
|
|
assert run is not None
|
|
payload = run_to_dict(run)
|
|
print(
|
|
"完成：{state} / {verdict}，循环 {loop_done}/{loop_target}，恢复 {recovery_count} 次。".format(
|
|
**payload
|
|
)
|
|
)
|
|
if bundle:
|
|
print("Evidence Bundle: {0}".format(bundle.uri))
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(prog="flashops")
|
|
parser.add_argument("command", choices=["init-db", "seed", "reset-db", "demo"])
|
|
args = parser.parse_args()
|
|
commands = {
|
|
"init-db": _init_db,
|
|
"seed": _seed,
|
|
"reset-db": _reset,
|
|
"demo": _demo,
|
|
}
|
|
asyncio.run(commands[args.command]())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|