clinic/netforce_clinic/models/setting.py

836 lines
31 KiB
Python

from netforce.model import Model, fields, get_model
from netforce.utils import get_file_path, get_data_path
from netforce.access import get_active_company, get_active_user, set_active_user
from netforce.database import get_connection
class ClinicSetting(Model):
_name="clinic.setting"
_string="Setting"
_fields={
"var_k": fields.Float("K"),
'signature': fields.File("Signature"),
'levels': fields.One2Many("clinic.setting.level","setting_id","Levels"),
'products': fields.One2Many("clinic.setting.product","setting_id","Products"),
'account_products': fields.One2Many("clinic.setting.account.product","setting_id","Account Products"),
'account_patients': fields.One2Many("clinic.setting.account.patient","setting_id","Account Patients"),
'shop_products': fields.One2Many("clinic.setting.shop.product","setting_id","Shop Products"),
'invoice_policies': fields.One2Many("clinic.setting.policy","setting_id","Invoice Policies"),
'cost_per_case': fields.Float("Cost Per Case"),
'company_id': fields.Many2One("company", 'Company'),
'period_id': fields.Many2One("clinic.period","Period"),
'waiting_approval': fields.Boolean("Waiting Approval"), # HD Case
'real_time': fields.Boolean("Real Time"), # HD Case
'patient_type_id': fields.Many2One("clinic.patient.type","Default Type"), # Import payment
'imp_patient_type_id': fields.Many2One("clinic.patient.type","Import UC Type"), # Import payment
'find_dlz': fields.Boolean("Find Dialyzer After Confirm Visit"), # Visit
'stock_journal_id': fields.Many2One("stock.journal","Default Journal"),
'auto_gen': fields.Boolean("Auto Gen"), # HD Case
'schd_from': fields.Date("From"),
'schd_to': fields.Date("To"),
'department_id': fields.Many2One("clinic.department","Department"),
'branch_id': fields.Many2One("clinic.branch","Branch"),
'shop_categs': fields.Many2Many("product.categ","Categs"),
'shop_type_id': fields.Many2One("clinic.patient.type","Patient Type"),
"cash_account_id": fields.Many2One("account.account","Cash Account",multi_company=True),
"income_account_id": fields.Many2One("account.account","Income Account",multi_company=True),
"import_account_id": fields.Many2One("account.account","Import Account",multi_company=True),
'helper_categ_id': fields.Many2One("clinic.staff.categ","Helper Category"),
'base_salary_day': fields.Float("Base Salary Day"),
}
_defaults={
"company_id": lambda *a: get_active_company(),
}
def onchange_line(self,context={}):
data=context['data']
path=context['path']
line=get_data_path(data,path,parent=True)
var_a=line['var_a'] or ''
var_b=line['var_b'] or ''
op=line['op'] or ''
line['formular']='%sX%s%s'%(var_a,op,var_b)
return data
def onchange_product(self,context={}):
data=context['data']
path=context['path']
line=get_data_path(data,path,parent=True)
product_id=line['product_id']
if product_id:
prod=get_model("product").browse(product_id)
uom=prod.uom_id
if not uom:
uom=get_model("uom").browse(1)
if not uom:
raise Exception("Not found uom 'Unit'")
line['uom_id']=uom.id
line['description']=prod.name
price=prod.sale_price or 0.0
line['price']=price
categ=prod.categ_id
if categ:
line['product_categ_id']=categ.id
qty=1
amt=qty*price
line['amount']=amt
return data
def onchange_ptype(self,context={}):
data=context['data']
path=context['path']
line=get_data_path(data,path,parent=True)
return data
def onchange_setting_line(self,context={}):
data=context['data']
path=context['path']
line=get_data_path(data,path,parent=True)
qty=line['qty'] or 0
price=line['price'] or 0
amt=qty*price
line['amount']=amt
return data
def schd_confirm(self,ids,context={}):
obj=self.browse(ids)[0]
if not obj.schd_from:
raise Exception("No date from")
if not obj.schd_to:
raise Exception("No date to")
dom=[]
dom.append(['date', '>=', obj.schd_from])
dom.append(['date', '<=', obj.schd_to])
for schd in get_model('clinic.schedule').search_browse(dom):
schd.confirm()
return True
def remove_rotation(self,ids,context={}):
staffs={}
for rt in get_model("clinic.staff.rotation").search_browse([]):
staff=rt.staff_id
if staff.name not in staffs:
staffs[staff.name]=[rt.id]
else:
staffs[staff.name].append(rt.id)
for st_name, vals in staffs.items():
for rt_id in vals:
rt=get_model("clinic.staff.rotation").browse(rt_id)
if not rt.level_id:
rt.delete()
print("Done!")
return
def update_cycle_item_level(self,ids,context={}):
for citem in get_model("clinic.cycle.item").search_browse([]):
for line in citem.lines:
nurse=line.nurse_id
if nurse.level_id:
line.write({
'level_id': nurse.level_id.id,
})
def run_script(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
for pt in get_model("clinic.patient").search_browse([]):
pt.partner_id.write({
'type': 'person',
'first_name': pt.first_name or "",
'last_name': pt.last_name or "",
})
for st in get_model("clinic.staff").search_browse([]):
st.partner_id.write({
'type': 'person',
'first_name': st.first_name or "",
'last_name': st.last_name or "",
})
print('Done!')
return
visits={}
for visit in get_model("clinic.visit").search_browse([]):
key='%s-%s'%(visit.visit_date, visit.patient_id.id)
if not key in visits.keys():
visits[key]=[]
hd_ids=[]
for hdcase in visit.hd_cases:
hd_ids.append(hdcase.id)
visits[key].append({
'id': visit.id,
'date': visit.visit_date,
'patient_id': visit.patient_id.id,
'state': visit.state,
'hd_ids': hd_ids,
})
for k, vs in visits.items():
if len(vs) > 1:
vids=[]
for v in vs:
if v['state']=='pending':
visit=get_model("clinic.visit").browse(v['id'])
for hdcase in visit.hd_cases:
hdcase.delete()
visit.delete()
print('deleted id ', v['id'], v['patient_id'], v['date'])
elif len(v['hd_ids']) < 1:
visit=get_model("clinic.visit").browse(v['id'])
if visit:
for hdcase in visit.hd_cases:
hdcase.delete()
visit.to_draft()
visit.delete()
print('deleted id ', v['id'], v['patient_id'], v['date'])
elif v['state'] in ('confirmed', 'cancelled'):
vids.append(v['id'])
print('need to delete id ', vids, vids[0])
#files=['SS3-1.csv','SS4-1.csv','SS4.csv']
#self.update_patient_file(files)
print("Done!")
def update_patient_file(self,files=[]):
import csv
for _file in files:
f=open("/tmp/%s"%_file,"r")
rd=csv.reader(f)
row1=next(rd)
headers=[]
for header in row1:
header='_'.join([x.lower() for x in header.split()])
headers.append(header)
lines=[]
for row in rd:
vals=dict(zip(headers,row))
lines.append(vals)
def get_dicts(model_name,dom=[]):
res={}
if not model_name:
return res
for sr in get_model(model_name).search_read(dom,['name']):
res[sr['name']]=sr['id']
return res
vasculars=get_dicts('clinic.vascular.access')
cycles=get_dicts('clinic.cycle')
titles=get_dicts('clinic.name.title')
default_title=None
for tname, tid in titles.items():
if tname=='No Title':
default_title=tid
break
types=get_dicts('clinic.patient.type')
dpts=get_dicts('clinic.department')
doctors=get_dicts(model_name='clinic.staff',dom=['type','=','doctor'])
days={
'Monday': 'mon',
'Tuesday': 'tue',
'Wednesday': 'wed',
'Thursday': 'thu',
'Friday': 'fri',
'Saturday': 'sat',
'Sunday': 'sun',
}
datas={}
olds=set({})
dbid2=0
dpt_id=None
for line in lines:
dbid=line['database_id']
if dbid not in olds:
dbid=line['database_id']
dbid2=dbid
else:
dbid=dbid2
olds.update({dbid})
fname=line['first_name']
lname=line['last_name']
dname=line['doctor']
cycle=line['cycles/cycle']
cycle=cycles.get(cycle,None)
day=line['cycles/day']
day=days.get(day)
dpt=line['department']
dpt=dpts.get(dpt,None)
if not dpt:
dpt=dpt_id
else:
dpt_id=dpt
bday=line['birthday'] or None
title=line['title']
title=titles.get(title,default_title) or None
idcard=line['id_card']
trt=line['trt']
type=line['type']
type=types.get(type,None)
valc=line['vascular_ac.']
valc=vasculars.get(valc)
hn=line['hn_number']
if not datas.get(dbid):
datas[dbid]={
'card_no': (idcard or "").replace(" ",""),
'number': hn,
'type_id': type,
'first_name': fname,
'last_name': lname,
'birthday': bday,
'title_id': title,
'trt_no': trt,
'doctor_id': doctors.get(dname),
'vascular_acc': valc,
'department_id': dpt,
}
print(" update patient data")
for patient_id, vals in datas.items():
if not patient_id:
continue
try:
m,d,y=(vals['birthday'] or "").split("/")
vals['birthday']='%s-%s-%s'%(y,m.zfill(2),d.zfill(2))
except:
pass
if not vals['card_no']:
vals['card_no']='/'
pvals=vals.copy()
get_model("clinic.patient").browse(int(patient_id)).write(pvals)
print("Done!")
def update_pcycle(self,files=[]):
import csv
for _file in files:
f=open("/tmp/%s"%_file,"r")
rd=csv.reader(f)
row1=next(rd)
headers=[]
for header in row1:
header='_'.join([x.lower() for x in header.split()])
headers.append(header)
lines=[]
for row in rd:
vals=dict(zip(headers,row))
lines.append(vals)
def get_dicts(model_name,dom=[]):
res={}
if not model_name:
return res
for sr in get_model(model_name).search_read(dom,['name']):
res[sr['name']]=sr['id']
return res
vasculars=get_dicts('clinic.vascular.access')
cycles=get_dicts('clinic.cycle')
titles=get_dicts('clinic.name.title')
default_title=None
for tname, tid in titles.items():
if tname=='No Title':
default_title=tid
break
types=get_dicts('clinic.patient.type')
dpts=get_dicts('clinic.department')
doctors=get_dicts(model_name='clinic.staff',dom=['type','=','doctor'])
days={
'Monday': 'mon',
'Tuesday': 'tue',
'Wednesday': 'wed',
'Thursday': 'thu',
'Friday': 'fri',
'Saturday': 'sat',
'Sunday': 'sun',
}
datas={}
olds=set({})
dbid2=0
dpt_id=None
for line in lines:
dbid=line['database_id']
if dbid not in olds:
dbid=line['database_id']
dbid2=dbid
else:
dbid=dbid2
olds.update({dbid})
fname=line['first_name']
lname=line['last_name']
dname=line['doctor']
cycle=line['cycles/cycle']
cycle=cycles.get(cycle,None)
day=line['cycles/day']
day=days.get(day)
dpt=line['department']
dpt=dpts.get(dpt,None)
if not dpt:
dpt=dpt_id
else:
dpt_id=dpt
bday=line['birthday'] or None
title=line['title']
title=titles.get(title,default_title) or None
idcard=line['id_card']
trt=line['trt']
type=line['type']
type=types.get(type,None)
valc=line['vascular_ac.']
valc=vasculars.get(valc)
hn=line['hn_number']
if not datas.get(dbid):
datas[dbid]={
'card_no': (idcard or "").replace(" ",""),
'number': hn,
'type_id': type,
'first_name': fname,
'last_name': lname,
'birthday': bday,
'title_id': title,
'trt_no': trt,
'doctor_id': doctors.get(dname),
'vascular_acc': valc,
'cycles': [],
}
if dbid and cycle and day:
datas[dbid]['cycles'].append({'department_id': dpt, 'cycle_id': cycle, 'day': day})
def del_pcycle(patient_id):
ids=get_model("clinic.patient.cycle").search([['patient_id','=',patient_id]])
get_model("clinic.patient.cycle").delete(ids)
print("pcycle del ", ids)
def create_pcycle(all_vals):
new_ids=[]
for vals in all_vals:
try:
id=get_model("clinic.patient.cycle").create(vals)
except Exception as e:
print("ERROR : ", e)
new_ids.append(id)
print("pcycle create ", new_ids)
print(" update patient data")
for patient_id, vals in datas.items():
if not patient_id:
continue
try:
m,d,y=(vals['birthday'] or "").split("/")
vals['birthday']='%s-%s-%s'%(y,m.zfill(2),d.zfill(2))
except:
pass
if not vals['card_no']:
vals['card_no']='/'
pvals=vals.copy()
del pvals['cycles']
get_model("clinic.patient").browse(int(patient_id)).write(pvals)
del_pcycle(patient_id)
all_vals=[]
for cycle_vals in vals['cycles']:
cycle_vals.update({'patient_id': patient_id})
all_vals.append(cycle_vals)
create_pcycle(all_vals)
print("Done!")
def update_staff_department(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
b={}
for dpt in get_model("clinic.department").search_browse([]):
branch=dpt.branch_id
if not branch.id in b:
b[branch.id]=[]
b[branch.id].append(dpt.id)
for staff in get_model("clinic.staff").search_browse([]):
if not staff.departments:
dpt=staff.department_id
branch=staff.branch_id
dpt_ids=[]
if dpt:
dpt_ids.append(dpt.id)
elif branch:
dpt_ids=b.get(branch.id)
staff.write({
'departments': [('set',dpt_ids)],
})
print(staff.name)
print("Done!")
return
def update_staff_level(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
# run below before above!
import csv
f=open("/tmp/update_rotation.csv","r")
rd=csv.reader(f)
headers=next(rd)
headers=[h.lower().replace(" ","") for h in headers]
lines=[]
for row in rd:
vals=dict(zip(headers,row))
lines.append(vals)
levels={}
for level in get_model("clinic.staff.level").search_read([]):
levels[level['name']]=level['id']
categories={}
for categ in get_model("clinic.staff.categ").search_read([]):
categories[categ['name']]=categ['id']
for line in lines:
db_id=int(line['databaseid'])
level=line['level']
level_id=levels.get(level,None)
catetory=line['category']
categ_id=categories.get(catetory,None)
rt=get_model('clinic.staff.rotation').browse(db_id)
rt.write({
'level_id': level_id,
'categ_id': categ_id,
})
#force update level, category store
for staff in get_model("clinic.staff").search_browse([]):
staff.write({
'note': ' ',
})
for citem in get_model("clinic.cycle.item").search_browse([]):
for line in citem.lines:
nurse=line.nurse_id
categ_id=nurse.categ_id
level_id=nurse.level_id
line.write({
'categ_id': categ_id.id,
'level_id': level_id.id,
})
print("Done!")
return
def remove_douplicate_visit(self,ids,context={}):
user_id=get_active_user()
set_active_user(1)
for pt in get_model("clinic.patient").search_browse([]):
dates=[]
print("="*50)
for visit in pt.visits:
if visit.state=='confirmed':
continue
if visit.visit_date in dates:
print(pt.name, ' delete ', visit.visit_date)
visit.to_draft()
visit.delete()
dates.append(visit.visit_date)
print("="*50)
set_active_user(user_id)
print("Done!")
return
def export_staff_rotation_csv(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
lines=[]
for rt in get_model("clinic.staff.rotation").search_browse([]):
staff=rt.staff_id
if staff.type!='nurse':
continue
level_name=''
if staff.level_id:
level_name=staff.level_id.name
lines.append({
'id': rt.id,
'name': staff.name,
'level': level_name,
'categ': staff.categ_id.name or "",
})
st="Database ID,Staff,Level,Category\n"
for line in sorted(lines,key=lambda x: x['name']):
st+="%s,%s,%s,%s\n"%(line['id'],line['name'],line['level'],line['categ'])
f=open("/tmp/rotation.csv","w")
f.write(st)
f.close()
print("Done!")
return
def update_departments(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
for st in get_model("clinic.staff").search_browse([]):
dpt_ids=set()
if st.departmet_id:
dpt_ids.update({st.departmet_id.id})
if st.type=='doctor':
for hdcase in st.hd_case_staffs:
dpt_ids.update({hdcase.department_id.id})
elif st.type=='nurse':
for citem in st.cycle_item_nurses:
dpt_ids.update({citem.department_id.id})
else:
continue
#FIXME
db=get_connection()
for dpt_id in dpt_ids:
if not dpt_id:
continue
sql="insert into m2m_clinic_department_clinic_staff values(%s,%s)"%(st.id, dpt_id)
# clinic_staff_id
# clinic_department_id
db.execute(sql)
#st.write({
#'departments': {'add': [1]},
#})
#break
print('update dpt %s'%st.name)
print("Done!")
def update_staff_contact(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
pids=[]
for st in get_model("clinic.staff").search_browse([]):
code=st.number
partner=st.partner_id
if partner:
partner.write({
'code': code,
'is_staff': True,
})
pids.append(partner.id)
for pt in get_model("clinic.patient").search_browse([]):
code=pt.hn_no or ""
partner=pt.partner_id
if partner:
partner.write({
'code': code,
'is_patient': True,
})
pids.append(partner.id)
for ptype in get_model('clinic.patient.type').search_browse([]):
partner=ptype.contact_id
if partner:
pids.append(partner.id)
#pids='(%s)'%(','.join([str(x) for x in pids]))
db=get_connection()
if pids:
res=db.query("select id, name from partner where id not in %s", tuple(pids))
pids=[r['id'] for r in res]
for pt in get_model("partner").browse(pids):
pt.write({
'active': False,
})
#for partner in get_model("partner").search_browse([]):
#pass
print("Done!")
def update_name(self,ids,context={}):
user_id=get_active_user()
if user_id !=1:
print("Only admin!!")
return
titles=dict([(t.name, t.id) for t in get_model("clinic.name.title").search_browse([])])
title_id=None
gtitle_id=title_id
for name, tid in titles.items():
if name=='No Title':
gtitle_id=tid
for model_name in ('clinic.staff', 'clinic.patient'):
for st in get_model(model_name).search_browse([]):
name=st.name
vals={}
if 'นายแพทย์' in name:
name=name.replace("นายแพทย์","")
tname='นายแพทย์'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'male',
})
elif 'นพ.' in name:
name=name.replace("นพ.","")
tname='นพ.'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'male',
})
elif 'นาย' in name:
name=name.replace("นาย","")
tname='นาย'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'male',
})
elif 'พันตำรวจเอก' in name:
name=name.replace("พันตำรวจเอก","")
tname='พันตำรวจเอก'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'male',
})
elif 'นางสาว' in name:
name=name.replace("นางสาว","")
tname='นางสาว'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'นาง' in name:
name=name.replace("นาง","")
tname='นาง'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'น.ส.' in name:
name=name.replace("น.ส.","")
tname='น.ส.'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'พญ..' in name:
name=name.replace("พญ..","")
tname='พญ..'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'แพทย์หญิง' in name:
name=name.replace("แพทย์หญิง","")
tname='แพทย์หญิง'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'ร้อยเอกหญิง' in name:
name=name.replace("ร้อยเอกหญิง","")
tname='ร้อยเอกหญิง'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'พันตรีหญิง' in name:
name=name.replace("พันตรีหญิง","")
tname='พันตรีหญิง'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
elif 'พันตำรวจโทหญิง' in name:
name=name.replace("พันตำรวจโทหญิง","")
tname='พันตำรวจโทหญิง'
title_id=titles.get(tname)
vals.update({
'title_id': title_id,
'gender': 'female',
})
else:
vals.update({
'first_name': name, #XXX
'title_id': gtitle_id,
#'gender': 'male',
})
names=name.split()
names=[n for n in names if n] # skip space
if len(names)==2:
vals.update({
'first_name':names[0],
'last_name': names[1],
})
else:
vals.update({
'first_name':name,
})
print(name)
st.write(vals)
if not st.gender:
st.write({
'gender': 'male',
})
print("Done!")
def reset_last_import(self,ids,context={}):
res=get_model("clinic.report.payment.matching").search_read([],['date'],order="date desc")
if res:
db=get_connection()
res1=res[0]
date1='%s 00:00:00'%res1['date']
date2='%s 23:59:59'%res1['date']
exp_ids=[x['id'] for x in db.query("select id from clinic_hd_case_expense where write_time>=%s and write_time<=%s",date1,date2)]
for exp in get_model("clinic.hd.case.expense").browse(exp_ids):
exp.write({
'state': 'waiting_matching',
'ok': False,
})
for inv in exp.invoices:
payment=inv.payment_id
if payment:
if payment.state !='draft':
payment.to_draft()
print("to draft payment ", payment.id)
print("Done")
def get_product_account(self,ids,prod_id=None,patient_type_id=None,pay_type='credit',context={}):
res={}
print("find ", prod_id, patient_type_id, pay_type)
dom=[['type','=',pay_type]]
for ac_prod in get_model("clinic.setting.account.product").search_browse(dom):
prod=ac_prod.product_id
ptype=ac_prod.patient_type_id
if patient_type_id==ptype.id and prod_id==prod.id:
print("found !")
res.update({
'ar_credit_id': ac_prod.ar_credit_id.id,
'ar_debit_id': ac_prod.ar_debit_id.id,
})
break
return res
def get_patient_hn(self,ids,context={}):
obj=self.browse(ids)[0]
res={
'hns':{},
'names': {},
}
for line in obj.account_patients:
pt=line.patient_id
name=line.partner_id.name or ""
vals={
'hn': pt.hn,
'pid': pt.id,
'card_no': pt.card_no,
'name': "%s%s"%(pt.first_name or "", pt.last_name or ""),
}
name=name.replace(" ","")
if name not in res['names'].keys():
res['names'][name]=vals
hn=line.hn
if hn not in res['hns'].keys():
res['hns'][hn]=vals
return res
ClinicSetting.register()