568 lines
26 KiB
Python
568 lines
26 KiB
Python
from django.db import models
|
||
from apps.system.models import CommonAModel, Dictionary, CommonBModel, CommonADModel, File, BaseModel
|
||
from rest_framework.exceptions import ParseError
|
||
from apps.utils.models import CommonBDModel
|
||
from collections import defaultdict, deque
|
||
from django.db.models import Q
|
||
from datetime import datetime, timedelta
|
||
from django.utils import timezone
|
||
|
||
class Process(CommonBModel):
|
||
"""
|
||
TN:工序
|
||
"""
|
||
PRO_PROD = 10
|
||
PRO_TEST = 20
|
||
|
||
|
||
PRO_NORMAL = 10
|
||
PRO_DIV = 20
|
||
PRO_MERGE = 30
|
||
|
||
name = models.CharField('工序名称', max_length=100)
|
||
type = models.PositiveSmallIntegerField("工序类型", default=PRO_PROD, choices=((PRO_PROD, '生产工序'), (PRO_TEST, '检验工序')))
|
||
mtype = models.PositiveSmallIntegerField("工序生产类型", default=PRO_NORMAL, choices=((PRO_NORMAL, '常规'), (PRO_DIV, '切分'), (PRO_MERGE, '合并')))
|
||
cate = models.CharField('大类', max_length=10, default='')
|
||
sort = models.PositiveSmallIntegerField('排序', default=1)
|
||
instruction = models.ForeignKey(
|
||
File, verbose_name='指导书', on_delete=models.SET_NULL, null=True, blank=True)
|
||
instruction_content = models.TextField('指导书内容', null=True, blank=True)
|
||
into_wm_mgroup = models.BooleanField('交接到工段', default=False)
|
||
store_notok = models.BooleanField('不合格品是否入库', default=False)
|
||
batch_append_equip = models.BooleanField('批号追加设备', default=False)
|
||
mlog_need_ticket = models.BooleanField('日志提交是否需要审批', default=False)
|
||
mstate_json = models.JSONField('中间状态', default=list, blank=True)
|
||
parent = models.ForeignKey('self', verbose_name='父工序', on_delete=models.CASCADE, null=True, blank=True)
|
||
number_to_batch = models.BooleanField('个号转批号', default=False)
|
||
wpr_number_rule = models.TextField("单个编号规则", null=True, blank=True)
|
||
|
||
class Meta:
|
||
verbose_name = '工序'
|
||
ordering = ['sort', 'create_time']
|
||
|
||
def get_canout_mat_ids(self):
|
||
"""获取可产出的materialIds
|
||
"""
|
||
return list(Route.objects.filter(process=self).values_list("material_out__id", flat=True).distinct())
|
||
|
||
def get_canin_mat_ids(self):
|
||
"""获取可输入的materialIds
|
||
"""
|
||
return list(RouteMat.objects.filter(route__process=self).values_list("material__id", flat=True).distinct()) + \
|
||
list(Route.objects.filter(process=self).values_list("material_in__id", flat=True).distinct())
|
||
|
||
# Create your models here.
|
||
class Material(CommonAModel):
|
||
"""TN:物料"""
|
||
MA_TYPE_BASE = 0
|
||
MA_TYPE_GOOD = 10
|
||
MA_TYPE_HALFGOOD = 20
|
||
MA_TYPE_MAINSO = 30
|
||
MA_TYPE_HELPSO = 40
|
||
MA_TYPE_TOOL = 50
|
||
MA_TYPE_HELPTOOL = 60
|
||
MA_TYPE_OFFICE = 70
|
||
|
||
type_choices = (
|
||
(MA_TYPE_BASE, '电/水/气'),
|
||
(MA_TYPE_GOOD, '成品'),
|
||
(MA_TYPE_HALFGOOD, '半成品'),
|
||
(MA_TYPE_MAINSO, '主要原料'),
|
||
(MA_TYPE_HELPSO, '辅助材料'),
|
||
(MA_TYPE_TOOL, '加工工具'),
|
||
(MA_TYPE_HELPTOOL, '辅助工装'),
|
||
(MA_TYPE_OFFICE, '办公用品')
|
||
)
|
||
|
||
MA_TRACKING_BATCH = 10
|
||
MA_TRACKING_SINGLE = 20
|
||
|
||
name = models.CharField('名称', max_length=50)
|
||
cate = models.CharField('大类', max_length=20, default='', blank=True)
|
||
number = models.CharField('编号', max_length=100, null=True, blank=True)
|
||
model = models.CharField(
|
||
'型号', max_length=100, null=True, blank=True)
|
||
specification = models.CharField(
|
||
'规格', max_length=100, null=True, blank=True)
|
||
code = models.CharField('标识', max_length=50, null=True, blank=True)
|
||
type = models.PositiveSmallIntegerField(
|
||
'物料类型', choices=type_choices, default=1, help_text=str(type_choices))
|
||
testitems = models.JSONField('检测项目', default=list, blank=True)
|
||
sort = models.FloatField('排序', default=1)
|
||
unit = models.CharField('基准计量单位', default='个', max_length=10)
|
||
tracking = models.PositiveSmallIntegerField("追踪方式", default=10,
|
||
choices=((MA_TRACKING_BATCH, '批次'),
|
||
(MA_TRACKING_SINGLE, '单件')))
|
||
count = models.DecimalField('总库存', max_digits=14, decimal_places=3, default=0)
|
||
count_mb = models.DecimalField('仓库库存', max_digits=14, decimal_places=3, default=0)
|
||
count_wm = models.DecimalField('车间库存', max_digits=14, decimal_places=3, default=0)
|
||
count_safe = models.DecimalField('安全库存数', max_digits=14, decimal_places=3, null=True, blank=True)
|
||
week_esitimate_consume = models.DecimalField('周消耗预估', max_digits=14, decimal_places=3, null=True, blank=True)
|
||
process = models.ForeignKey(
|
||
Process, verbose_name='所用工序', on_delete=models.CASCADE, null=True, blank=True)
|
||
parent = models.ForeignKey(
|
||
'self', null=True, blank=True, on_delete=models.SET_NULL, verbose_name='父物料')
|
||
is_hidden = models.BooleanField('是否隐藏', default=False)
|
||
is_assemb = models.BooleanField('是否组合件', default=False)
|
||
need_route = models.BooleanField('是否需要定义工艺路线', default=False)
|
||
components = models.JSONField('组件', default=dict, null=False, blank=True)
|
||
brothers = models.JSONField('兄弟件', default=list, null=False, blank=True)
|
||
unit_price = models.DecimalField('单价', max_digits=14, decimal_places=2, null=True, blank=True)
|
||
into_wm = models.BooleanField('是否进入车间库存', default=True)
|
||
bin_number_main = models.CharField('主库位号', max_length=50, null=True, blank=True)
|
||
img = models.TextField('图片', null=True, blank=True)
|
||
|
||
class Meta:
|
||
verbose_name = '物料表'
|
||
ordering = ['sort', '-create_time']
|
||
|
||
def __str__(self):
|
||
return f'{self.name}|{self.specification if self.specification else ""}|{self.model if self.model else ""}|{self.process.name if self.process else ""}'
|
||
|
||
|
||
class Shift(CommonBModel):
|
||
"""TN:班次
|
||
"""
|
||
name = models.CharField('名称', max_length=50)
|
||
rule = models.CharField('所属规则', max_length=10, default='默认')
|
||
start_time_o = models.TimeField('开始时间')
|
||
end_time_o = models.TimeField('结束时间')
|
||
sort = models.PositiveSmallIntegerField('排序', default=1)
|
||
|
||
class Meta:
|
||
verbose_name = '班次'
|
||
|
||
class Srule(CommonBDModel):
|
||
"""
|
||
TN:班组规则
|
||
"""
|
||
rule = models.JSONField('排班规则', default=list, blank=True)
|
||
|
||
|
||
class Team(CommonBModel):
|
||
"""TN:班组, belong_dept为所属车间
|
||
"""
|
||
name = models.CharField('名称', max_length=50)
|
||
leader = models.ForeignKey(
|
||
'system.user', verbose_name='班长', on_delete=models.CASCADE)
|
||
|
||
|
||
class Mgroup(CommonBModel):
|
||
"""TN:测点集
|
||
"""
|
||
M_SELF = 10
|
||
M_OTHER = 20
|
||
name = models.CharField('名称', max_length=50)
|
||
code = models.CharField('标识', max_length=50, null=True, blank=True)
|
||
cate = models.CharField(
|
||
'分类', max_length=50, default='section', help_text='section/other') # section是工段
|
||
mtype = models.PositiveSmallIntegerField("工段类型", default=10, help_text='10:自产 20:外协')
|
||
shift_rule = models.CharField('班次规则', max_length=10, default='默认')
|
||
process = models.ForeignKey(
|
||
Process, verbose_name='工序', on_delete=models.SET_NULL, null=True, blank=True)
|
||
product = models.ForeignKey(
|
||
Material, verbose_name='主要产品', on_delete=models.SET_NULL, null=True, blank=True, related_name='mgroup_product')
|
||
product_cost = models.ForeignKey(
|
||
Material, verbose_name='主要产品(成本计算)', on_delete=models.SET_NULL, null=True, blank=True, related_name='mgroup_product_cost')
|
||
input_materials = models.JSONField(
|
||
'直接材料', default=list, blank=True, help_text='material的ID列表')
|
||
test_materials = models.JSONField(
|
||
'检测材料', default=list, blank=True, help_text='material的ID列表')
|
||
batch_append_code = models.BooleanField('批号追加工段标识', default=False)
|
||
sort = models.PositiveSmallIntegerField('排序', default=1)
|
||
need_enm = models.BooleanField('是否进行能源监测', default=True)
|
||
is_running = models.BooleanField('是否正常运行', default=False)
|
||
|
||
class Meta:
|
||
verbose_name = '测点集'
|
||
ordering = ['sort', '-create_time']
|
||
|
||
def __str__(self) -> str:
|
||
return self.name
|
||
|
||
def get_shift(self, w_s_time:datetime):
|
||
# 如果没有时区信息，使用默认时区（东八区）
|
||
if not timezone.is_aware(w_s_time):
|
||
w_s_time = timezone.make_aware(w_s_time)
|
||
else:
|
||
w_s_time = timezone.localtime(w_s_time)
|
||
|
||
shifts = Shift.objects.filter(rule=self.shift_rule).order_by('sort')
|
||
if not shifts:
|
||
raise ParseError(f"工段{self.name}未配置班次")
|
||
# 处理跨天班次的情况
|
||
for shift in shifts:
|
||
# 如果开始时间小于结束时间，表示班次在同一天内
|
||
if shift.start_time_o < shift.end_time_o:
|
||
if shift.start_time_o <= w_s_time.time() < shift.end_time_o:
|
||
return w_s_time.date(), shift
|
||
else: # 班次跨天（如夜班从当天晚上到次日凌晨）
|
||
if w_s_time.time() >= shift.start_time_o or w_s_time.time() < shift.end_time_o:
|
||
# 如果当前时间在开始时间之后，属于当天
|
||
if w_s_time.time() >= shift.start_time_o:
|
||
return w_s_time.date(), shift
|
||
# 如果当前时间在结束时间之前，属于前一天
|
||
else:
|
||
return (w_s_time - timedelta(days=1)).date(), shift
|
||
# return w_s_time.date(), None
|
||
|
||
|
||
class TeamMember(BaseModel):
|
||
team = models.ForeignKey(Team, verbose_name='关联班组',
|
||
on_delete=models.CASCADE)
|
||
user = models.ForeignKey(
|
||
'system.user', verbose_name='成员', on_delete=models.CASCADE)
|
||
mgroup = models.ForeignKey(
|
||
Mgroup, verbose_name='所在工段', on_delete=models.CASCADE)
|
||
post = models.ForeignKey('system.post', verbose_name='岗位',
|
||
on_delete=models.SET_NULL, null=True, blank=True)
|
||
|
||
|
||
class Goal(CommonADModel):
|
||
"""TN:目标
|
||
"""
|
||
mgroup = models.ForeignKey(
|
||
Mgroup, verbose_name='关联工段', on_delete=models.CASCADE, null=True, blank=True)
|
||
year = models.PositiveSmallIntegerField('年')
|
||
goal_cate = models.ForeignKey(
|
||
Dictionary, verbose_name='目标种类', on_delete=models.CASCADE)
|
||
goal_val = models.FloatField('全年目标值')
|
||
goal_val_1 = models.FloatField('1月份目标值')
|
||
goal_val_2 = models.FloatField('2月份目标值')
|
||
goal_val_3 = models.FloatField('3月份目标值')
|
||
goal_val_4 = models.FloatField('4月份目标值')
|
||
goal_val_5 = models.FloatField('5月份目标值')
|
||
goal_val_6 = models.FloatField('6月份目标值')
|
||
goal_val_7 = models.FloatField('7月份目标值')
|
||
goal_val_8 = models.FloatField('8月份目标值')
|
||
goal_val_9 = models.FloatField('9月份目标值')
|
||
goal_val_10 = models.FloatField('10月份目标值')
|
||
goal_val_11 = models.FloatField('11月份目标值')
|
||
goal_val_12 = models.FloatField('12月份目标值')
|
||
|
||
class Meta:
|
||
unique_together = ("mgroup", "year", "goal_cate")
|
||
|
||
|
||
class RoutePack(CommonADModel):
|
||
"""
|
||
TN:加工工艺
|
||
"""
|
||
RP_S_CREATE = 10
|
||
RP_S_AUDIT = 20
|
||
RP_S_CONFIRM = 30
|
||
RP_STATE = (
|
||
(RP_S_CREATE, '创建中'),
|
||
(RP_S_AUDIT, '审批中'),
|
||
(RP_S_CONFIRM, '已确认'),
|
||
)
|
||
state = models.PositiveSmallIntegerField('状态', default=10, choices=RP_STATE)
|
||
name = models.CharField('名称', max_length=100)
|
||
material = models.ForeignKey(
|
||
Material, verbose_name='产品', on_delete=models.CASCADE, null=True, blank=True)
|
||
ticket = models.ForeignKey('wf.ticket', verbose_name='关联工单',
|
||
on_delete=models.SET_NULL, related_name='routepack_ticket', null=True, blank=True, db_constraint=False)
|
||
document = models.ForeignKey("system.file", verbose_name='工艺文件', on_delete=models.SET_NULL, null=True, blank=True, db_constraint=False)
|
||
gjson = models.JSONField('子图结构', default=dict, blank=True)
|
||
|
||
def validate_and_return_gjson(self, raise_exception=True):
|
||
"""
|
||
校验所有子图，返回结构化数据：
|
||
{
|
||
final_material_out_id1: {
|
||
"routes": [1, 2, ...], # 子图关联的Route ID
|
||
"meta": {
|
||
"is_valid": True,
|
||
"errors": [],
|
||
"levels": {material_id: 层级深度}
|
||
}
|
||
},
|
||
final_material_out_id2: { ... }
|
||
}
|
||
"""
|
||
all_routes = Route.objects.filter(routepack=self).order_by('sort', 'process__sort', 'create_time')
|
||
result = {}
|
||
|
||
# 1. 构建正向和反向图（material_id作为节点）
|
||
forward_graph = defaultdict(list) # out_id -> [in_ids]
|
||
reverse_graph = defaultdict(list) # in_id -> [out_ids]
|
||
route_edges = defaultdict(list) # (in_id, out_id) -> [Route]
|
||
|
||
for route in all_routes:
|
||
if not (route.material_in and route.material_out):
|
||
raise ParseError(f"Route(ID:{route.id}) 缺失material_in或material_out")
|
||
in_id, out_id = route.material_in.id, route.material_out.id
|
||
forward_graph[out_id].append(in_id)
|
||
reverse_graph[in_id].append(out_id)
|
||
route_edges[(in_id, out_id)].append(route.id)
|
||
|
||
# 2. 识别所有最终节点（未被任何in_id引用的out_id）
|
||
all_out_ids = {r.material_out.id for r in all_routes}
|
||
all_in_ids = {r.material_in.id for r in all_routes}
|
||
final_ids = all_out_ids - all_in_ids
|
||
|
||
if not final_ids:
|
||
raise ParseError("未找到最终产品节点")
|
||
|
||
# 3. 为每个最终产品构建子图
|
||
for final_id in final_ids:
|
||
# BFS逆向遍历所有上游节点
|
||
visited = set()
|
||
queue = deque([final_id])
|
||
|
||
while queue:
|
||
current_out_id = queue.popleft()
|
||
if current_out_id in visited:
|
||
continue
|
||
visited.add(current_out_id)
|
||
|
||
# 将当前节点的所有上游in_id加入队列
|
||
for in_id in forward_graph.get(current_out_id, []):
|
||
if in_id not in visited:
|
||
queue.append(in_id)
|
||
|
||
# 收集所有关联Route（只要out_id在子图内）
|
||
subgraph_route_ids = []
|
||
for route in all_routes:
|
||
if route.material_out.id in visited:
|
||
subgraph_route_ids.append(route.id)
|
||
|
||
result[final_id] = {"routes": subgraph_route_ids}
|
||
return result
|
||
|
||
def get_gjson(self, need_update=False, final_material_id=None):
|
||
if not self.gjson or need_update:
|
||
self.gjson = self.validate_and_return_gjson()
|
||
self.save()
|
||
return self.gjson.get(final_material_id, {}) if final_material_id else self.gjson
|
||
|
||
def get_dags(self):
|
||
gjson = self.get_gjson()
|
||
materialIdList = []
|
||
routeIdList = []
|
||
for final_material_id, data in gjson.items():
|
||
materialIdList.append(final_material_id)
|
||
routeIdList.extend(data['routes'])
|
||
|
||
# 获取所有相关路由，并确保关联了material_in和material_out
|
||
route_qs = Route.objects.filter(id__in=routeIdList).select_related("material_in", "material_out", "process")
|
||
|
||
# 收集所有相关物料ID，并过滤掉None值
|
||
matids1 = [mid for mid in route_qs.values_list("material_in__id", flat=True) if mid is not None]
|
||
matids2 = [mid for mid in route_qs.values_list("material_out__id", flat=True) if mid is not None]
|
||
materialIdList.extend(matids1)
|
||
materialIdList.extend(matids2)
|
||
|
||
# 构建路由字典，添加空值检查
|
||
route_dict = {}
|
||
for r in route_qs:
|
||
if r.material_in and r.material_out: # 只有两个物料都存在时才添加
|
||
route_dict[r.id] = {
|
||
"label": r.process.name if r.process else "",
|
||
"source": r.material_in.id,
|
||
"target": r.material_out.id,
|
||
"id": r.id
|
||
}
|
||
|
||
# 获取所有物料信息
|
||
mat_qs = Material.objects.filter(id__in=materialIdList).order_by("process__sort", "create_time")
|
||
mat_dict = {mat.id: {"id": mat.id, "label": str(mat), "shape": "rect"} for mat in mat_qs}
|
||
|
||
res = {}
|
||
for final_material_id, data in gjson.items():
|
||
# 确保最终物料存在于mat_dict中
|
||
if final_material_id not in mat_dict:
|
||
continue
|
||
|
||
item = {"name": mat_dict[final_material_id]["label"]}
|
||
edges = []
|
||
nodes_set = set()
|
||
|
||
for route_id in data['routes']:
|
||
# 只处理存在于route_dict中的路由
|
||
if route_id in route_dict:
|
||
edges.append(route_dict[route_id])
|
||
nodes_set.update([route_dict[route_id]['source'], route_dict[route_id]['target']])
|
||
|
||
item['edges'] = edges
|
||
# 只添加存在于mat_dict中的节点
|
||
item['nodes'] = [mat_dict[node_id] for node_id in nodes_set if node_id in mat_dict]
|
||
res[final_material_id] = item
|
||
|
||
return res
|
||
|
||
def get_final_material_ids(self):
|
||
return list(self.get_gjson().keys())
|
||
class Route(CommonADModel):
|
||
"""
|
||
TN:加工路线
|
||
"""
|
||
routepack = models.ForeignKey(RoutePack, verbose_name='关联路线包', on_delete=models.CASCADE, null=True, blank=True)
|
||
material = models.ForeignKey(
|
||
Material, verbose_name='关联产品', on_delete=models.CASCADE, null=True, blank=True)
|
||
process = models.ForeignKey(
|
||
Process, verbose_name='工序', on_delete=models.CASCADE, null=True, blank=True, related_name="route_p")
|
||
mgroup = models.ForeignKey(Mgroup, verbose_name='指定工段', on_delete=models.CASCADE,
|
||
null=True, blank=True, related_name='route_mgroup')
|
||
sort = models.PositiveSmallIntegerField('顺序', default=1)
|
||
is_autotask = models.BooleanField('是否自动排产', default=False)
|
||
material_in = models.ForeignKey(
|
||
Material, verbose_name='主要输入物料', on_delete=models.CASCADE, related_name='route_material_in', null=True, blank=True)
|
||
material_out = models.ForeignKey(
|
||
Material, verbose_name='主要输出物料', on_delete=models.CASCADE, related_name='route_material_out', null=True, blank=True)
|
||
is_count_utask = models.BooleanField('是否主任务统计', default=False)
|
||
out_rate = models.FloatField('出材率', default=100, blank=True)
|
||
div_number = models.PositiveSmallIntegerField('切分数量', default=1, blank=True)
|
||
hour_work = models.FloatField('工时', null=True, blank=True)
|
||
batch_bind = models.BooleanField('是否绑定批次', default=True)
|
||
materials = models.ManyToManyField(Material, verbose_name='关联辅助物料', related_name="route_materials",
|
||
through="mtm.routemat", blank=True)
|
||
parent = models.ForeignKey('self', verbose_name='上级路线', on_delete=models.CASCADE, null=True, blank=True, related_name="route_parent")
|
||
params_json = models.JSONField('工艺参数', default=dict, blank=True)
|
||
from_route = models.ForeignKey('self', verbose_name='来源路线', on_delete=models.SET_NULL, null=True, blank=True, related_name="route_f")
|
||
|
||
def __str__(self):
|
||
x = ""
|
||
if self.material_in:
|
||
x = x + str(self.material_in) + "->"
|
||
if self.material_out:
|
||
x = x + str(self.material_out)
|
||
return x
|
||
|
||
@staticmethod
|
||
def get_routes(material: Material=None, routepack:RoutePack=None, routeIds=None):
|
||
"""
|
||
TN:返回工艺路线带车间(不关联工艺包)
|
||
"""
|
||
if material:
|
||
kwargs = {'material': material, 'routepack__isnull': True}
|
||
rqs = Route.objects.filter(
|
||
**kwargs).order_by('sort', 'process__sort', 'create_time')
|
||
# if not rq.exists():
|
||
# raise ParseError('未配置工艺路线')
|
||
# if rq.first().material_in is None or rq.last().material_out is None or rq.last().material_out != rq.last().material:
|
||
# raise ParseError('首步缺少输入/最后一步缺少输出')
|
||
# if not rq.filter(is_count_utask=True).exists():
|
||
# raise ParseError('未指定统计步骤')
|
||
elif routepack:
|
||
rqs = Route.objects.filter(routepack=routepack).order_by('sort', 'process__sort', 'create_time')
|
||
elif routeIds:
|
||
rqs = Route.objects.filter(id__in=routeIds).order_by('sort', 'process__sort', 'create_time')
|
||
return rqs
|
||
|
||
@classmethod
|
||
def validate_dag(cls, final_material_out:Material, rqs):
|
||
"""
|
||
TN:校验工艺路线是否正确：
|
||
- 所有 Route 必须有 material_in 和 material_out
|
||
- 所有路径最终指向给定的 final_material_out
|
||
- 无循环依赖
|
||
- 所有物料必须能到达 final_material_out
|
||
"""
|
||
# 1. 检查字段完整性
|
||
invalid_routes = rqs.filter(
|
||
models.Q(material_in__isnull=True) |
|
||
models.Q(material_out__isnull=True)
|
||
)
|
||
if invalid_routes.exists():
|
||
raise ParseError("存在 Route 记录缺失 material_in 或 material_out")
|
||
|
||
# 2. 构建依赖图（使用material_id作为键）
|
||
graph = defaultdict(list) # {material_out_id: [material_in_id]}
|
||
reverse_graph = defaultdict(list) # {material_in_id: [material_out_id]}
|
||
all_material_ids = set()
|
||
|
||
for route in rqs.all():
|
||
out_id = route.material_out.id
|
||
in_id = route.material_in.id
|
||
|
||
graph[out_id].append(in_id)
|
||
reverse_graph[in_id].append(out_id)
|
||
all_material_ids.update({in_id, out_id})
|
||
|
||
# 3. 检查final_material_out是否是终点
|
||
final_id = final_material_out.id
|
||
if final_id in reverse_graph:
|
||
# raise ParseError(
|
||
# f"最终物料 {final_material_out.name}(ID:{final_id}) 不能作为任何Route的输入"
|
||
# )
|
||
raise ParseError("最终物料不可作为输入")
|
||
|
||
# 4. BFS检查路径可达性（使用material_id操作）
|
||
visited = set()
|
||
queue = deque([final_id])
|
||
|
||
while queue:
|
||
current_id = queue.popleft()
|
||
if current_id in visited:
|
||
continue
|
||
visited.add(current_id)
|
||
for in_id in graph.get(current_id, []):
|
||
queue.append(in_id)
|
||
|
||
# 5. 检查未到达的物料
|
||
unreachable_ids = all_material_ids - visited
|
||
if unreachable_ids:
|
||
# unreachable_materials = Material.objects.filter(id__in=unreachable_ids).values_list('name', flat=True)
|
||
# raise ParseError(
|
||
# f"以下物料无法到达最终物料: {list(unreachable_materials)}"
|
||
# )
|
||
raise ParseError("存在无法到达的节点")
|
||
|
||
# 6. DFS检查循环依赖
|
||
visited_cycle = set()
|
||
path_cycle = set()
|
||
|
||
def has_cycle(material_id):
|
||
if material_id in path_cycle:
|
||
return True
|
||
if material_id in visited_cycle:
|
||
return False
|
||
visited_cycle.add(material_id)
|
||
path_cycle.add(material_id)
|
||
for out_id in reverse_graph.get(material_id, []):
|
||
if has_cycle(out_id):
|
||
return True
|
||
path_cycle.remove(material_id)
|
||
return False
|
||
|
||
for material_id in all_material_ids:
|
||
if has_cycle(material_id):
|
||
# cycle_material = Material.objects.get(id=material_id)
|
||
# raise ParseError(f"循环依赖涉及物料: {cycle_material.name}(ID:{material_id})")
|
||
raise ParseError('存在循环依赖')
|
||
|
||
return True
|
||
|
||
@classmethod
|
||
def get_dag(cls, rqs):
|
||
# 收集所有相关批次和边
|
||
nodes_set = set()
|
||
edges = []
|
||
|
||
for rq in rqs:
|
||
if rq.material_in and rq.material_out:
|
||
source = rq.material_in.id
|
||
target = rq.material_out.id
|
||
nodes_set.update([source, target])
|
||
edges.append({
|
||
'source': source,
|
||
'target': target,
|
||
'label': rq.process.name,
|
||
'id': rq.id
|
||
})
|
||
# 将批次号排序
|
||
nodes_qs = Material.objects.filter(id__in=nodes_set).order_by("process__sort", "create_time")
|
||
|
||
# 构建节点数据，默认使用'rect'形状
|
||
nodes = [{
|
||
'id': item.id,
|
||
'label': str(item),
|
||
'shape': 'rect' # 可根据业务需求调整形状
|
||
} for item in nodes_qs]
|
||
|
||
return {'nodes': nodes, 'edges': edges}
|
||
class RouteMat(BaseModel):
|
||
"""TN:工艺路线辅助物料"""
|
||
route = models.ForeignKey(Route, verbose_name='关联路线', on_delete=models.CASCADE, related_name="routemat_route")
|
||
material = models.ForeignKey(Material, verbose_name='辅助物料', on_delete=models.CASCADE) |