"""Production department schedule: 09:00-18:00, fixed weekends off, no holidays."""
|
import sys,json
|
from pathlib import Path
|
from datetime import date,datetime,time,timedelta
|
P=Path(__file__).parent
|
sys.stdout.reconfigure(encoding='utf8')
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
assert not (P/'result.json').exists(),'Already filled; verify rather than repeat.'
|
START=date(2026,9,1);END=date(2026,10,14)
|
holidays={date(2026,9,d):'中秋节' for d in range(25,28)}|{date(2026,10,d):'国庆节' for d in range(1,8)}
|
days=[START+timedelta(days=i) for i in range((END-START).days+1)]
|
workdays=[d for d in days if d.weekday()<5 and d not in holidays]
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',autocommit=False,cursorclass=pymysql.cursors.DictCursor)
|
result={'plans':[],'members':[],'holiday_ids':[],'schedule_ids':[],'source':'https://www.beijing.gov.cn/cs/gncs/zcwj/202603/t20260327_4568275.html','fixed_weekends_off':True,'start':'2026-09-01','end':'2026-10-14','shift':'09:00–18:00'}
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'")
|
q.execute('SELECT DATABASE() db,NOW() now,GET_LOCK(%s,10) locked',('production-schedule-20261014',));state=q.fetchone();assert state['db']=='mom-hm' and state['locked']==1;now=state['now'];assert now.date()>=END
|
tabs=['mes_cal_plan','mes_cal_plan_shift','mes_cal_plan_team','mes_cal_team','mes_cal_team_member','mes_cal_team_shift','mes_cal_holiday','system_dept','system_users']
|
before={};schema={}
|
for t in tabs:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id'+(' FOR UPDATE' if t.startswith('mes_cal_') else ''));before[t]=q.fetchall()
|
q.execute('SHOW COLUMNS FROM '+t);schema[t]={x['Field'] for x in q.fetchall()}
|
q.execute("SELECT * FROM system_dict_data WHERE dict_type='mes_cal_calendar_type' ORDER BY id FOR UPDATE");dicts=q.fetchall()
|
backup={'state':state,'tables':before,'dicts':dicts}
|
(P/'locked-before.json').write_text(json.dumps(backup,ensure_ascii=False,default=str,indent=2),encoding='utf8')
|
production={d['id'] for d in before['system_dept'] if d['status']==0 and d['name'] in ['生产部','生产部门']};assert production
|
while True:
|
extended=production|{d['id'] for d in before['system_dept'] if d['status']==0 and d['parent_id'] in production}
|
if extended==production:break
|
production=extended
|
users=[u for u in before['system_users'] if u['status']==0 and u['dept_id'] in production];assert users
|
actor=next(u for u in users if u['id']==244)
|
q.execute('SELECT COUNT(*) n FROM system_user_role WHERE user_id=244 AND role_id=166 AND deleted=0');assert q.fetchone()['n']>0
|
def insert(t,v):
|
v={**v,'creator':'244','updater':'244','create_time':now,'update_time':now,'deleted':0};assert set(v)<=schema[t],(t,set(v)-schema[t])
|
q.execute('INSERT INTO '+t+' ('+','.join('`'+k+'`' for k in v)+') VALUES ('+','.join(['%s']*len(v))+')',list(v.values()));return q.lastrowid
|
active=[d for d in dicts if d['deleted'] in (0,b'\x00') and d['status']==0]
|
candidates=[d for d in active if d['label']=='生产']
|
if candidates:calendar_type=int(candidates[0]['value']);result['new_calendar_type']=None
|
else:
|
calendar_type=max([int(d['value']) for d in dicts if str(d['value']).isdigit()] or [0])+1
|
q.execute('INSERT INTO system_dict_data (sort,label,value,dict_type,status,color_type,remark,creator,updater,create_time,update_time,deleted) VALUES (%s,%s,%s,%s,0,%s,%s,%s,%s,%s,%s,0)',(calendar_type,'生产',str(calendar_type),'mes_cal_calendar_type','primary','生产部门白班排班','244','244',now,now));result['new_calendar_type']=q.lastrowid
|
result['calendar_type']=calendar_type
|
assert not any(t['code']=='SCBB-2026' for t in before['mes_cal_team']), 'Production team already exists; inspect before appending.'
|
team=insert('mes_cal_team',{'code':'SCBB-2026','name':'生产白班组','calendar_type':calendar_type,'remark':'生产部门人员；09:00–18:00；固定双休，节假日休息'})
|
result['team_id']=team;result['department_ids']=sorted(production)
|
for u in users:
|
id=insert('mes_cal_team_member',{'team_id':team,'user_id':u['id']});result['members'].append({'id':id,'user_id':u['id'],'name':u['nickname'],'dept_id':u['dept_id']})
|
# Explicit holiday overrides ensure native confirm/generate also excludes them.
|
holiday_dates={x['day'].date():x for x in before['mes_cal_holiday']}
|
assert len(holiday_dates)==len(before['mes_cal_holiday']),'Duplicate existing holiday dates'
|
for day,label in holidays.items():
|
if day in holiday_dates:
|
assert holiday_dates[day]['type']==2,('Conflicting holiday',str(day));id=holiday_dates[day]['id']
|
else:id=insert('mes_cal_holiday',{'day':datetime.combine(day,time.min),'type':2,'remark':label})
|
result['holiday_ids'].append(id)
|
# Fixed weekends remain off even for the national adjusted working weekends.
|
assert not any(h['type']==1 and START<=h['day'].date()<=END and h['day'].weekday()>=5 for h in before['mes_cal_holiday']), 'Existing weekend working overrides conflict with fixed weekends.'
|
for month,start,end in [('202609',START,date(2026,9,30)),('202610',date(2026,10,1),END)]:
|
code='SC-'+month
|
assert not any(x['code']==code for x in before['mes_cal_plan']),('Plan already exists',code)
|
pid=insert('mes_cal_plan',{'code':code,'name':'生产部门'+('9月' if month=='202609' else '10月1日至14日')+'排班','calendar_type':calendar_type,'start_date':datetime.combine(start,time.min),'end_date':datetime.combine(end,time.min),'shift_type':1,'shift_method':2,'shift_count':1,'status':0,'exclude_weekend':1,'remark':'09:00–18:00；固定双休，法定节假日不排班'})
|
sid=insert('mes_cal_plan_shift',{'plan_id':pid,'sort':1,'name':'白班','start_time':'09:00','end_time':'18:00'})
|
link=insert('mes_cal_plan_team',{'plan_id':pid,'team_id':team})
|
dates=[d for d in workdays if start<=d<=end]
|
for day in dates:
|
id=insert('mes_cal_team_shift',{'plan_id':pid,'team_id':team,'shift_id':sid,'day':datetime.combine(day,time.min),'sort':1});result['schedule_ids'].append(id)
|
q.execute('UPDATE mes_cal_plan SET status=1,updater=%s,update_time=%s WHERE id=%s',('244',now,pid));assert q.rowcount==1
|
result['plans'].append({'id':pid,'code':code,'shift_id':sid,'plan_team_id':link,'days':[str(d) for d in dates]})
|
# Check no prior business records or staff have been modified.
|
for t,rows in before.items():
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');after={x['id']:x for x in q.fetchall()}
|
for old in rows:assert after[old['id']]==old,('Existing data changed',t,old['id'])
|
if not t.startswith('mes_cal_'):assert len(after)==len(rows)
|
assert len(workdays)==26 and len(result['schedule_ids'])==26 and len(users)==45
|
c.commit()
|
# Post-commit independent read of the generated relationships and all dates.
|
q.execute('SELECT * FROM mes_cal_team_shift WHERE team_id=%s AND deleted=0 ORDER BY day',(team,));shifts=q.fetchall()
|
assert [s['day'].date() for s in shifts]==workdays and len({s['day'] for s in shifts})==len(workdays)
|
q.execute('SELECT * FROM mes_cal_team_member WHERE team_id=%s AND deleted=0',(team,));members=q.fetchall();assert {m['user_id'] for m in members}=={u['id'] for u in users} and len(members)==len(users)
|
for p in result['plans']:
|
q.execute('SELECT * FROM mes_cal_plan WHERE id=%s',(p['id'],));p_row=q.fetchone();assert p_row['status']==1 and p_row['exclude_weekend']==1
|
q.execute('SELECT * FROM mes_cal_plan_shift WHERE id=%s',(p['shift_id'],));s_row=q.fetchone();assert s_row['start_time']=='09:00' and s_row['end_time']=='18:00'
|
assert all(s['shift_id']==p['shift_id'] for s in shifts if s['plan_id']==p['id'])
|
result.update(committed=True,verified=True,at=str(now),workdays=26,users=45,person_days=26*45,excluded_days=[str(d) for d in days if d not in workdays])
|
(P/'result.json').write_text(json.dumps(result,ensure_ascii=False,default=str,indent=2),encoding='utf8')
|
print(json.dumps({k:result[k] for k in ['committed','verified','at','start','end','shift','team_id','users','workdays','person_days','calendar_type']},ensure_ascii=False))
|
except Exception:c.rollback();raise
|
finally:
|
with c.cursor() as q:q.execute('SELECT RELEASE_LOCK(%s)',('production-schedule-20261014',))
|
c.close()
|
