You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
157 lines
4.2 KiB
157 lines
4.2 KiB
""" 命令行脚本, 基于typer来完成 """
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import time
|
|
import logging
|
|
import typer
|
|
import glob
|
|
import json
|
|
|
|
from tortoise import Tortoise, run_async
|
|
|
|
import commands
|
|
from conf import setting, register_mysql
|
|
|
|
app = typer.Typer()
|
|
|
|
|
|
def enable_debug_sql():
|
|
import logging, sys
|
|
|
|
fmt = logging.Formatter(
|
|
fmt="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
sh = logging.StreamHandler(sys.stdout)
|
|
sh.setLevel(logging.DEBUG)
|
|
sh.setFormatter(fmt)
|
|
|
|
# will print debug sql
|
|
logger_db_client = logging.getLogger("tortoise.db_client")
|
|
logger_db_client.setLevel(logging.DEBUG)
|
|
print("--------" * 3)
|
|
|
|
|
|
if setting.SQL_DEBUG_MODE:
|
|
enable_debug_sql()
|
|
|
|
|
|
async def init():
|
|
"""初始化数据库"""
|
|
await Tortoise.init(
|
|
config={
|
|
"connections": {"default": setting.DB_URL},
|
|
"apps": {"rms": {"models": ["models"], "default_connection": "default"}},
|
|
'use_tz': False,
|
|
'timezone': setting.TIMEZONE,
|
|
}
|
|
)
|
|
|
|
|
|
async def close():
|
|
"""关闭数据库连接"""
|
|
await Tortoise.close_connections()
|
|
|
|
|
|
async def main(callback):
|
|
"""主体处理函数"""
|
|
await init()
|
|
await callback()
|
|
await close()
|
|
|
|
|
|
@app.command()
|
|
def init_generate_schemas():
|
|
"""初始化数据库中的所有表信息(危险操作)"""
|
|
|
|
async def _inner():
|
|
# await Tortoise.generate_schemas()
|
|
pass
|
|
|
|
run_async(main(_inner))
|
|
|
|
|
|
@app.command()
|
|
def delete_some_drugs():
|
|
"""删除某个试剂在数据库中的记录"""
|
|
from models import Drug, DrugUseLog
|
|
|
|
async def _inner():
|
|
rfids = ["11719", "11720"]
|
|
for _rfid in rfids:
|
|
_drug = await Drug.filter(rfid=_rfid).first()
|
|
if not _drug:
|
|
continue
|
|
_drug_use_logs = await DrugUseLog.filter(drug_id=_drug.id).all()
|
|
print("Rfid:", _rfid)
|
|
for _drug_use_log in _drug_use_logs:
|
|
print("DrugUseLog:", _drug_use_log.id, _drug.id)
|
|
await _drug_use_log.delete()
|
|
await _drug.delete()
|
|
|
|
run_async(main(_inner))
|
|
|
|
|
|
@app.command()
|
|
def delete_cabinet(
|
|
terminal_id: str = typer.Option(default=None, help="终端ID"),
|
|
cabinet_id: str = typer.Option(default=None, help="柜体ID"),
|
|
force: int = typer.Option(default=0, help="是否强制删除: 1-强制, 0-需确认"),
|
|
):
|
|
"""根据传入的参数值cabinet_id或terminal_id进行柜体的删除操作, 注意, 这是一个非常危险的操作, 请不要随意还行, 否则后果自负"""
|
|
from models import Drawer, DrawerBoard
|
|
|
|
async def _inner():
|
|
from commands.clear_one_cabinet_command import clear_cabinet
|
|
|
|
if not force:
|
|
confirmation = input("你正在进行一个非常危险的操作, 请确认是否继续(Y/N, 默认为N): ") or 'N'
|
|
confirmation = confirmation.upper()
|
|
if confirmation != 'Y':
|
|
sys.exit(0)
|
|
|
|
if not terminal_id and not cabinet_id:
|
|
print("--柜体ID和终端ID都为空, 请检查输入参数")
|
|
sys.exit(-1)
|
|
|
|
await clear_cabinet(cabinet_id, terminal_id=terminal_id)
|
|
|
|
run_async(main(_inner))
|
|
|
|
|
|
@app.command()
|
|
def add_terminal(
|
|
config_file: str = typer.Option(help="配置文件路径"),
|
|
force: int = typer.Option(default=0, help="是否强制删除: 1-强制, 0-需确认"),
|
|
):
|
|
"""根据传入的配置文件路径, 自动解析该配置文件并往数据库中添加终端/柜体/抽屉等信息"""
|
|
|
|
async def _inner():
|
|
from commands.init_command import command_add_terminal_by_config
|
|
|
|
if not force:
|
|
confirmation = input("你正在进行一个非常危险的操作, 请确认是否继续(Y/N, 默认为N): ") or 'N'
|
|
confirmation = confirmation.upper()
|
|
if confirmation != 'Y':
|
|
sys.exit(0)
|
|
|
|
await command_add_terminal_by_config(config_file)
|
|
|
|
run_async(main(_inner))
|
|
|
|
|
|
@app.command()
|
|
def temp_test():
|
|
"""临时更新数据操作, 可能涉及比较危险的操作"""
|
|
|
|
async def _inner():
|
|
pass
|
|
|
|
run_async(main(_inner))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app()
|