cews开发注意事项

一.gitlab部署步骤

merge request 从个人分支 -> main分支
merge request main分支 -> stage分支
点CI/CD 查看

二.docker compose 生成django镜像慢

1. dockerfile里添加国内镜像源地址

apt 添加镜像源

RUN echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm main contrib non-free non-free-firmware" > /etc/apt/sources.list && \
    echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-updates main contrib non-free non-free-firmware" >> /etc/apt/sources.list && \
    echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ bookworm-backports main contrib non-free non-free-firmware" >> /etc/apt/sources.list && \
    echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian-security bookworm-security main contrib non-free non-free-firmware" >> /etc/apt/sources.list

pip 添加镜像源

RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple && \
  pip config set install.trusted-host mirrors.aliyun.com

项目启动相关配置

local.py

# 打开范围权限
ALLOWED_HOSTS = ["*"]

# 取消debug配置
# INSTALLED_APPS += ["debug_toolbar"]  # noqa: F405
# MIDDLEWARE += ["debug_toolbar.middleware.DebugToolbarMiddleware"]  # noqa: F405

# 打开跨域
CORS_ALLOW_CREDENTIALS = True
CORS_ORIGIN_ALLOW_ALL = True
# CORS_ORIGIN_WHITELIST = (
#     'http://127.0.0.1:8080',
# )
CORS_ALLOWED_ORIGINS_REGEXES=[
    r'^http://.*?$',
]
# CORS_ORIGIN_REGEXES_WHITELIST = (
#         r'^http://.*?$',
# )
CORS_ALLOW_METHODS = (
    'DELETE',
    'GET',
    'OPTIONS',
    'PATCH',
    'POST',
    'PUT',
    'VIEW',
)

CORS_ALLOW_HEADERS = (
    'XMLHttpRequest',
    'X_FILENAME',
    'accept-encoding',
    'authorization',
    'content-type',
    'dnt',
    'origin',
    'user-agent',
    'x-csrftoken',
    'x-requested-with',
    'Pragma',
)

gnsuser.py

self.__connection = self.testConnection()
改为
self.__connection = True

axios.ts

const api = axios.create({ baseURL: 'http://192.168.1.80:8001' });

.env

# # General
# # ------------------------------------------------------------------------------
USE_DOCKER=yes
IPYTHONDIR=/app/.ipython
# # Redis
# # ------------------------------------------------------------------------------
REDIS_URL=redis://127.0.0.1:6379/0

# Celery
# ------------------------------------------------------------------------------

# Flower
CELERY_FLOWER_USER=debug
CELERY_FLOWER_PASSWORD=debug
CELERY_BROKER_URL=redis://127.0.0.1:6379/0


MONGO_DATABASE_NAME=eform
MONGO_HOST=127.0.0.1
MONGO_PORT=27017
MONGO_INITDB_ROOT_USERNAME=debug
MONGO_INITDB_ROOT_PASSWORD=debug
MONGO_DATABASE_NAME=eform
ME_CONFIG_MONGODB_URL=mongodb://${MONGO_INITDB_ROOT_USERNAME}:${MONGO_INITDB_ROOT_PASSWORD}@mongo:27017/
# PostgreSQL
# ------------------------------------------------------------------------------
# POSTGRES_HOST=postgres
# POSTGRES_PORT=5432
# POSTGRES_DB=web_backend
# POSTGRES_USER=debug
POSTGRES_HOST=127.0.0.1
POSTGRES_PORT=5432
POSTGRES_DB=CEWS_PRDO_TEST
POSTGRES_USER=inhouse
POSTGRES_PASSWORD=12345678
DATABASE_URL="postgres://inhouse:12345678@127.0.0.1:5432/CEWS_PRDO_TEST"

三.表结构

1. 已`RepairOrderPart`表和`RepairOrderRepairDetails`为基准(与history, 和maintenance一致)再查出RepairOrder表为准

    repair_order_part_ids = set(RepairOrderPart.objects.filter(deleted=False).values_list("repair_order_id", flat=True))

    # V1
    details = RepairOrderRepairDetails.objects.select_related('repair_order').filter(deleted=False).filter(repair_order_id__in=repair_order_part_ids)

    # V2
    RepairOrder.objects.filter(pk__in=repair_order_part_ids,deleted=False)

2. 查询出maintence或者history的数据(原来的人使用的v1方法, 但是实际我使用v2方法返回的也是一样的)

:V1

    # rop_ids = set(RepairOrderPart.objects.filter(deleted=False).values_list("repair_order_id", flat=True))
    # ro_queryset = RepairOrder.objects.only("id").filter(id__in=rop_ids, deleted=False)
    # ro_count = ro_queryset.count()
    # rop_ids_list = []
    # for ro in ro_queryset:
    #     r_o_p = ro.primary_part
    #     rop_ids_list.append(r_o_p.id)
    # rop_ids_list.sort()
    # print(f"count1_list: {rop_ids_list}")

:V2

    rop_ids = set(RepairOrderPart.objects.filter(deleted=False).values_list("repair_order_id", flat=True))
    ro_ids = RepairOrder.objects.filter(id__in=rop_ids, deleted=False).values_list("id", flat=True)
    rop_ids_list = list(RepairOrderPart.objects.filter(
        repair_order_id__in=ro_ids,
        parent_part__isnull=True
    ).values_list("id", flat=True))
    rop_ids_list.sort()
    print(rop_ids_list)

:V3

    # # 获取所有未删除的 repair_order_id
    # rop_ids = set(RepairOrderPart.objects.filter(deleted=False).values_list("repair_order_id", flat=True))
    # # 获取所有与这些 repair_order_id 相关的 RepairOrder
    # ro_ids = list(RepairOrder.objects.filter(id__in=rop_ids, deleted=False).values_list("id", flat=True))
    # # 创建一个子查询，选择每个 repair_order_id 的第一条记录
    # subquery = RepairOrderPart.objects.filter(
    #     repair_order_id=OuterRef('repair_order_id'),
    #     parent_part__isnull=True
    # ).order_by('id').values('id')[:1]
    # # 获取每个 repair_order_id 对应的第一条 RepairOrderPart 的 id
    # rop_ids_list = list(set(RepairOrderPart.objects.filter(
    #     repair_order_id__in=ro_ids
    # ).annotate(
    #     first_id=Subquery(subquery)
    # ).values_list('first_id', flat=True)))
    # rop_ids_list.sort()
    # print(rop_ids_list)

测试

def not_fitement_level(level1, level2, level3) -> dict[str, int]:
    """计算上一次已经check_in, 下一次还没check_out, 并且没有装车的lv4数量"""
    rop_ids = set(RepairOrderPart.objects.filter(deleted=False).values_list("repair_order_id", flat=True))
    ro_ids = RepairOrder.objects.filter(id__in=rop_ids, location="", deleted=False).values_list("id", flat=True)

    query = Q(parts__lv1=level1, parts__lv2=level2, parts__lv3=level3, parts__last_check_out__isnull=False)

    parts_ids = set(RepairOrderPart.objects.filter(repair_order_id__in=ro_ids,parent_part__isnull=True).filter(query).values_list("parts_id", flat=True))

    lv4_count_list = list(Parts.objects.filter(id__in=parts_ids, last_order__location="").values("lv4").annotate(count=Count("id")))

    ### CSV TEST
    csv_data_list = list(Parts.objects.filter(id__in=parts_ids, last_order__location="").values_list("id","stock_code", "serial_no"))#.values("lv4").annotate(count=Count("id"))
    print(lv4_count_list)
    with open("/home/chf/python_objs/CEWS/test1.csv", "w") as file:
        writer = csv.writer(file)
        writer.writerows(csv_data_list)

    lv4_count_dict = {}
    lv4_count_dict["category"] = [f"{level1} / {level2} / {level3}"]
    legend = []
    for lv4_count in lv4_count_list:
        lv4_name = lv4_count["lv4"]
        lv4_count_dict[lv4_name] = [lv4_count["count"]]
        if lv4_name not in legend:
            legend.append(lv4_name)
    lv4_count_dict["legend"] = legend
    return lv4_count_dict

def get_critical_incident_repair_summary(year: int = datetime.now().year, month: int = datetime.now().month):
    status_list = [RepairOrder.STATUS.CHECKED_IN, RepairOrder.STATUS.IN_PROCESS, RepairOrder.STATUS.WAITING_FOR_ENDORSEMENT, RepairOrder.STATUS.WAITING_FOR_SIC, RepairOrder.STATUS.WAITING_FOR_JOB_COMPLETED, RepairOrder.STATUS.COMPLETE, RepairOrder.STATUS.FORCE_COMPLETE]

    data = []

    def location_type_process(location_type:str):
        if location_type:
            if "M-Train" in location_type:
                return "M-Train"
            elif "C-Train" in location_type:
                return "C-Train"
            elif "K-Train" in location_type:
                return "K-Train"
            elif "CAF-Train" in location_type:
                return "CAF-Train"
        return "Total"

    def data_serialize(repair_order_obj, repair_order_repair_details_obj):
        repair_order: RepairOrderPart = repair_order_obj.primary_part
        result_type = ""
        if repair_order_repair_details_obj.test_result in ["FF", "NAF"]:
            result_type = repair_order_repair_details_obj.test_result
        elif repair_order_repair_details_obj.test_result == "Within spec." and repair_order_repair_details_obj.maintenance_type == "NFF":
            result_type = repair_order_repair_details_obj.maintenance_type
        else:
            result_type = "under_repair"

        return {
            "stock_code": repair_order.stock_code,
            "serial_no": repair_order.serial_no,
            "location_type": location_type_process(repair_order_obj.location_type),
            "mtce_type": repair_order_obj.mtce_type,
            "test_result": repair_order_repair_details_obj.test_result, 
            "maintenance_type": repair_order_repair_details_obj.maintenance_type,
            "check_in": format_datetime(repair_order_obj.check_in),
            "check_out": format_datetime(repair_order_obj.check_out),
            "result_type": result_type,
            "lv1": repair_order.parts.lv1,
            "lv2": repair_order.parts.lv2,
            "lv3": repair_order.parts.lv3,
        }
    repair_order_ids = set(RepairOrderPart.objects.filter(status__in=status_list,repair_order__check_in__month=month,repair_order__mtce_type=RepairOrder.MTCE_TYPE.CM,repair_order__check_in__year=year, deleted=False).values_list("repair_order_id", flat=True))
    for repair_order_id in repair_order_ids:
        repair_order_queryset = RepairOrder.objects.only("id", "location_type", "check_in", "check_out").filter(pk=repair_order_id, deleted=False) 
        if repair_order_queryset.exists():
            repair_order_repair_details_queryset = RepairOrderRepairDetails.objects.only("id","test_result", "maintenance_type").filter(repair_order_id=repair_order_id)
            if repair_order_repair_details_queryset.exists():
                repair_order_obj = repair_order_queryset.first()
                repair_order_repair_details_obj = repair_order_repair_details_queryset.first()
                data.append(data_serialize(repair_order_obj, repair_order_repair_details_obj))


    return {"table": critical_incident_repair_summary_data_process(data), "data": data}


def critical_incident_repair_summary_data_process(data):
    result_types = ["FF", "NFF", "under_repair", "NAF"]

    # Total代表全部类型的车, 包括"M-Train", "C-Train", "K-Train", "CAF-Train"
    all_train = ["Total", "M-Train", "C-Train", "K-Train", "CAF-Train"] 
    data_group = {
        "Total":[],
        "M-Train":[],
        "C-Train":[],
        "K-Train":[],
        "CAF-Train":[]
    }

    # 初始化 result 字典
    result = {rtype: [0] * len(all_train) for rtype in result_types}

    result["train_type"] = all_train

    # 统计出现次数
    counts = defaultdict(lambda: defaultdict(int))

    # 进行数据分组
    for entry in data:
        location = entry["location_type"]
        if location != "Total":
            data_group[location].append(entry)
        data_group["Total"].append(entry)


    # 数据过滤
    filter_data = data_group["M-Train"] + data_group["C-Train"] + data_group["K-Train"] + data_group["CAF-Train"] 
    total_data = data_group["Total"]

    # 单独处理total全部车辆的
    for entry in total_data:
        result_type = entry["result_type"]
        counts[result_type]["Total"] += 1

    # 再处理特定车辆的
    for entry in filter_data:
        location = entry["location_type"]
        result_type = entry["result_type"]
        if result_type != "Total":
            counts[result_type][location] += 1

    # 填充 result 字典
    for result_type in result_types:
        for i, location in enumerate(all_train):
            result[result_type][i] = counts[result_type].get(location, 0)
    return result

CHF Blog ~

ChenBlog

[ cews开发注意事项 ]

一.gitlab部署步骤

二.docker compose 生成django镜像慢

1. dockerfile里添加国内镜像源地址

项目启动相关配置

三.表结构

1. 已`RepairOrderPart`表和`RepairOrderRepairDetails`为基准(与history, 和maintenance一致)再查出RepairOrder表为准

2. 查询出maintence或者history的数据(原来的人使用的v1方法, 但是实际我使用v2方法返回的也是一样的)

:V1

:V2

:V3

测试

您还没有登录，请您登录后发表评论。

一.gitlab部署步骤

二.docker compose 生成django镜像慢

1. dockerfile里添加国内镜像源地址

项目启动相关配置

三.表结构

1. 已RepairOrderPart表和RepairOrderRepairDetails为基准(与history, 和maintenance一致)再查出RepairOrder表为准

2. 查询出maintence或者history的数据(原来的人使用的v1方法, 但是实际我使用v2方法返回的也是一样的)

:V1

:V2

:V3

测试

您还没有登录，请您登录后发表评论。

1. 已`RepairOrderPart`表和`RepairOrderRepairDetails`为基准(与history, 和maintenance一致)再查出RepairOrder表为准