201 lines
7.6 KiB
Python
201 lines
7.6 KiB
Python
"""命令行管理工具。
|
||
|
||
用法（在 backend/ 目录下执行）::
|
||
|
||
uv run python -m app.cli create-admin --username admin
|
||
|
||
密码优先**交互式输入**而非命令行参数：命令行参数会进入 shell 历史，
|
||
在多用户机器上还能被 `ps` 看到。`--password` 仅为自动化脚本保留。
|
||
"""
|
||
|
||
import argparse
|
||
import getpass
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import select
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.constants import PASSWORD_MAX_BYTES, PASSWORD_MIN_BYTES, ROLE_ADMIN
|
||
from app.core.db import dispose_engine, get_session_factory
|
||
from app.core.security import hash_password
|
||
from app.core.tplus_db import dispose_tplus_engine, get_tplus_session_factory
|
||
from app.models import Role, User, UserRole
|
||
from app.repositories.tplus_repository import get_batch
|
||
from app.services.tplus.sync_service import TPlusSyncService
|
||
from app.services.tplus.worker import run_once
|
||
|
||
|
||
def _prompt_password() -> str:
|
||
"""交互式读取并二次确认密码，循环直到符合策略。"""
|
||
while True:
|
||
password = getpass.getpass(
|
||
f"请输入密码（{PASSWORD_MIN_BYTES}-{PASSWORD_MAX_BYTES} 字节）: "
|
||
)
|
||
length = len(password.encode("utf-8"))
|
||
if length < PASSWORD_MIN_BYTES:
|
||
print(f" 密码不足 {PASSWORD_MIN_BYTES} 字节，请重新输入。")
|
||
continue
|
||
if length > PASSWORD_MAX_BYTES:
|
||
print(f" 密码超过 {PASSWORD_MAX_BYTES} 字节上限（中文一字占 3 字节），请重新输入。")
|
||
continue
|
||
if password != getpass.getpass("请再次输入以确认: "):
|
||
print(" 两次输入不一致，请重新输入。")
|
||
continue
|
||
return password
|
||
|
||
|
||
def _create_admin(args: argparse.Namespace) -> int:
|
||
"""幂等创建管理员账号。
|
||
|
||
已存在时**不覆盖密码** —— 意外重跑脚本不应把已有账号的密码改掉。
|
||
"""
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
existing = session.execute(
|
||
select(User).where(User.username == args.username)
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
print(f"用户 {args.username!r} 已存在（id={existing.id}），未做任何修改。")
|
||
return 0
|
||
|
||
role = session.execute(
|
||
select(Role).where(Role.role_code == ROLE_ADMIN)
|
||
).scalar_one_or_none()
|
||
if role is None:
|
||
print(
|
||
"错误：未找到 admin 角色，请先执行数据库迁移（uv run alembic upgrade head）。",
|
||
file=sys.stderr,
|
||
)
|
||
return 1
|
||
|
||
password = args.password or _prompt_password()
|
||
user = User(
|
||
username=args.username,
|
||
password=hash_password(password),
|
||
real_name=args.real_name,
|
||
status=1,
|
||
)
|
||
session.add(user)
|
||
session.flush() # 取得自增 id 后才能建关联
|
||
session.add(UserRole(user_id=user.id, role_id=role.id))
|
||
session.commit()
|
||
|
||
print(f"已创建管理员 {args.username!r}（id={user.id}），角色：{role.role_name}。")
|
||
print("首次登录后请立即修改密码。")
|
||
return 0
|
||
|
||
|
||
def _stage_tplus(args: argparse.Namespace) -> int:
|
||
"""读取 T+ 源库并写入 MySQL staging。"""
|
||
target_factory = get_session_factory()
|
||
source_factory = get_tplus_session_factory()
|
||
watermark = datetime.fromisoformat(args.watermark) if args.watermark else None
|
||
with source_factory() as source_session, target_factory() as target_session:
|
||
batch = TPlusSyncService(source_session, target_session).run(
|
||
entity_types=args.entity,
|
||
watermark=watermark,
|
||
batch_size=args.batch_size,
|
||
)
|
||
print(
|
||
f"T+ staging 完成：批次={batch.batch_no} 状态={batch.status} "
|
||
f"源记录={batch.source_record_count} 新增={batch.staged_record_count} "
|
||
f"失败={batch.failed_record_count}"
|
||
)
|
||
return 0 if batch.status == "SUCCEEDED" else 1
|
||
|
||
|
||
def _tplus_worker(args: argparse.Namespace) -> int:
|
||
"""独立消费持久化任务；HTTP 进程不承担长时间读取。"""
|
||
if args.connection_file:
|
||
settings = get_settings()
|
||
settings.t_plus_enabled, settings.t_plus_connection_file = True, args.connection_file
|
||
if not 2 <= args.poll_seconds <= 60:
|
||
print("轮询间隔必须在 2 到 60 秒之间", file=sys.stderr)
|
||
return 1
|
||
factory = get_session_factory()
|
||
try:
|
||
while True:
|
||
with factory() as session:
|
||
batch_no = run_once(session)
|
||
batch = get_batch(session, batch_no) if batch_no else None
|
||
status = batch.status if batch else None
|
||
if batch_no:
|
||
print(f"任务完成：{batch_no} 状态={status}", flush=True)
|
||
if not args.watch:
|
||
return 0 if status in (None, "SUCCEEDED") else 1
|
||
if batch_no is None or status != "SUCCEEDED":
|
||
time.sleep(args.poll_seconds)
|
||
except KeyboardInterrupt:
|
||
print("T+ 读取服务已停止")
|
||
return 130
|
||
except Exception as exc:
|
||
print(f"T+ 读取服务停止（{type(exc).__name__}），请检查运行环境", file=sys.stderr)
|
||
return 1
|
||
|
||
|
||
def main() -> int:
|
||
"""命令行入口。"""
|
||
parser = argparse.ArgumentParser(
|
||
prog="app.cli", description="衍水集团营销人员工资管理系统 - 命令行工具"
|
||
)
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
|
||
p_admin = sub.add_parser("create-admin", help="创建初始管理员账号（幂等）")
|
||
p_admin.add_argument("--username", default="admin", help="登录用户名，默认 admin")
|
||
p_admin.add_argument("--real-name", default="系统管理员", help="真实姓名")
|
||
p_admin.add_argument(
|
||
"--password",
|
||
default=None,
|
||
help="密码。不传则交互式输入（推荐）——命令行传参会留在 shell 历史中",
|
||
)
|
||
|
||
p_tplus = sub.add_parser("tplus-stage", help="读取 T+ 源库并写入原始 staging")
|
||
p_tplus.add_argument(
|
||
"--entity",
|
||
action="append",
|
||
choices=[
|
||
"personnel",
|
||
"products",
|
||
"customers",
|
||
"businesses",
|
||
"shipments",
|
||
"receipts",
|
||
"business_items",
|
||
"shipment_items",
|
||
"receipt_items",
|
||
],
|
||
required=True,
|
||
help="实体类型，可重复指定",
|
||
)
|
||
p_tplus.add_argument(
|
||
"--watermark",
|
||
default=None,
|
||
help="增量起始时间，ISO 8601 格式；不传表示全量读取",
|
||
)
|
||
p_tplus.add_argument("--batch-size", type=int, default=500, help="每次从源库读取的行数")
|
||
|
||
p_worker = sub.add_parser("tplus-worker", help="执行已排队的 T+ 读取或关联核验")
|
||
modes = p_worker.add_mutually_exclusive_group()
|
||
modes.add_argument("--once", action="store_true", help="执行一个任务后退出（默认）")
|
||
modes.add_argument("--watch", action="store_true", help="持续消费手动排队任务")
|
||
p_worker.add_argument("--connection-file", default=None, help="显式指定项目外只读连接 JSON")
|
||
p_worker.add_argument("--poll-seconds", type=int, default=3, help="空闲轮询间隔，2–60 秒")
|
||
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
if args.command == "create-admin":
|
||
return _create_admin(args)
|
||
if args.command == "tplus-worker":
|
||
return _tplus_worker(args)
|
||
return _stage_tplus(args)
|
||
finally:
|
||
dispose_engine()
|
||
dispose_tplus_engine()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|