zlwu
fa5efa9629
Reuse assignment business columns and keyword searches in the four statistics drill-down lists. Add a statistics-permission read endpoint sharing tenant-scoped detail queries without granting assignment access. Related to issue #13. Independent static review completed. User self-test arrangement: regression cases authored, no tests or builds executed.
198 lines
10 KiB
Python
198 lines
10 KiB
Python
"""项目业务明细检索；统计读取与归属维护分别校验菜单权限。"""
|
|
from collections import defaultdict
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import and_, func, or_
|
|
|
|
from app.models.models import (
|
|
BizChannelServiceOrder, BizCollectionHospital,
|
|
BizPilotPatient, BizProtocolCollectionInfo, BizProtocolPartyCustomer, BizProtocolSample, SysUser,
|
|
)
|
|
from app.models.research_donation import (
|
|
BizResearchDonationCollection, BizResearchDonationParty, BizResearchDonationSample,
|
|
)
|
|
from app.models.project_management import Project, ProjectTeam
|
|
from app.services import project_management as projects
|
|
|
|
|
|
PARTY_MODELS = {
|
|
'cell_storage': (BizProtocolPartyCustomer, BizProtocolCollectionInfo, 'collection_protocol_id'),
|
|
'research_donation': (BizResearchDonationParty, BizResearchDonationCollection, 'research_donation_id'),
|
|
}
|
|
SAMPLE_MODELS = {
|
|
'cell_storage': BizProtocolSample,
|
|
'research_donation': BizResearchDonationSample,
|
|
}
|
|
|
|
|
|
def keyword_condition(db, tenant, kind, keyword):
|
|
"""用 EXISTS 检索子表，避免多条患者/采集信息放大主单计数。"""
|
|
model, number, _ = projects.BUSINESS[kind]
|
|
conditions = [getattr(model, number).contains(keyword, autoescape=True)]
|
|
if kind in PARTY_MODELS:
|
|
party, collection, parent_key = PARTY_MODELS[kind]
|
|
sample = SAMPLE_MODELS[kind]
|
|
conditions.append(projects.scoped(db, sample, tenant).filter(
|
|
getattr(sample, parent_key) == model.id,
|
|
sample.sample_code.contains(keyword, autoescape=True),
|
|
).exists())
|
|
conditions.append(projects.scoped(db, party, tenant).filter(
|
|
getattr(party, parent_key) == model.id,
|
|
or_(party.mother_name.contains(keyword, autoescape=True),
|
|
party.mother_phone.contains(keyword, autoescape=True)),
|
|
).exists())
|
|
hospitals = projects.scoped(db, BizCollectionHospital, tenant).filter(
|
|
BizCollectionHospital.hospital_name.contains(keyword, autoescape=True),
|
|
).with_entities(BizCollectionHospital.id)
|
|
sales_users = projects.scoped(db, SysUser, tenant).filter(
|
|
SysUser.nick_name.contains(keyword, autoescape=True),
|
|
).with_entities(SysUser.id)
|
|
conditions.append(projects.scoped(db, collection, tenant).filter(
|
|
getattr(collection, parent_key) == model.id,
|
|
or_(collection.hospital_id.in_(hospitals), collection.sales_user_id.in_(sales_users)),
|
|
).exists())
|
|
elif kind == 'pilot_service':
|
|
conditions.extend(column.contains(keyword, autoescape=True) for column in (
|
|
model.application_name, model.channel_name,
|
|
))
|
|
conditions.append(projects.scoped(db, BizPilotPatient, tenant).filter(
|
|
BizPilotPatient.pilot_application_id == model.id,
|
|
or_(BizPilotPatient.patient_name.contains(keyword, autoescape=True),
|
|
BizPilotPatient.phone_no.contains(keyword, autoescape=True)),
|
|
).exists())
|
|
conditions.append(projects.scoped(db, BizChannelServiceOrder, tenant).filter(
|
|
BizChannelServiceOrder.id == model.channel_service_order_id,
|
|
BizChannelServiceOrder.order_name.contains(keyword, autoescape=True),
|
|
).exists())
|
|
else:
|
|
conditions.extend(column.contains(keyword, autoescape=True) for column in (
|
|
model.project_title, model.customer_name, model.customer_contact, model.customer_phone,
|
|
))
|
|
return or_(*conditions)
|
|
|
|
|
|
def child_rows(db, tenant, model, parent_key, ids, fields):
|
|
"""仅批量读取当前页所需字段；稳定排序保留多条有效子记录。"""
|
|
result = defaultdict(list)
|
|
if ids:
|
|
parent = getattr(model, parent_key)
|
|
rows = projects.scoped(db, model, tenant).filter(parent.in_(ids)).with_entities(
|
|
parent, *(getattr(model, field) for field in fields),
|
|
).order_by(model.create_time, model.id).all()
|
|
for row in rows:
|
|
result[getattr(row, parent_key)].append(row)
|
|
return result
|
|
|
|
|
|
def joined_values(rows, field):
|
|
return '、'.join(dict.fromkeys(str(getattr(row, field)) for row in rows if getattr(row, field)))
|
|
|
|
|
|
def project_metadata(db, tenant, rows):
|
|
ids = {row.project_id for row in rows if row.project_id}
|
|
if not ids:
|
|
return {}
|
|
values = projects.scoped(db, Project, tenant).outerjoin(ProjectTeam, and_(
|
|
ProjectTeam.id == Project.team_id, ProjectTeam.tenant_no == tenant, ProjectTeam.del_flag == 0,
|
|
)).filter(Project.id.in_(ids)).with_entities(
|
|
Project.id, Project.project_name, Project.team_id, ProjectTeam.team_name,
|
|
).all()
|
|
return {row.id: dict(projectName=row.project_name, teamId=row.team_id, teamName=row.team_name)
|
|
for row in values}
|
|
|
|
|
|
def business_details(db, tenant, kind, rows):
|
|
ids = [row.id for row in rows]
|
|
result = {row.id: {} for row in rows}
|
|
if not ids:
|
|
return result
|
|
if kind in PARTY_MODELS:
|
|
party, collection, parent_key = PARTY_MODELS[kind]
|
|
parties = child_rows(db, tenant, party, parent_key, ids, ('mother_name', 'mother_phone'))
|
|
samples = child_rows(db, tenant, SAMPLE_MODELS[kind], parent_key, ids, ('sample_code',))
|
|
collections = child_rows(db, tenant, collection, parent_key, ids,
|
|
('hospital_id', 'sales_user_id', 'sample_collect_time'))
|
|
all_collections = [item for values in collections.values() for item in values]
|
|
hospital_ids = {item.hospital_id for item in all_collections if item.hospital_id}
|
|
sales_ids = {item.sales_user_id for item in all_collections if item.sales_user_id}
|
|
hospitals = dict(projects.scoped(db, BizCollectionHospital, tenant).filter(
|
|
BizCollectionHospital.id.in_(hospital_ids),
|
|
).with_entities(BizCollectionHospital.id, BizCollectionHospital.hospital_name).all()) if hospital_ids else {}
|
|
sales_users = dict(projects.scoped(db, SysUser, tenant).filter(
|
|
SysUser.id.in_(sales_ids),
|
|
).with_entities(SysUser.id, SysUser.nick_name).all()) if sales_ids else {}
|
|
for row in rows:
|
|
info = collections[row.id]
|
|
details = dict(
|
|
sampleCodes=joined_values(samples[row.id], 'sample_code'),
|
|
motherName=joined_values(parties[row.id], 'mother_name'),
|
|
motherPhone=joined_values(parties[row.id], 'mother_phone'),
|
|
hospitalName='、'.join(dict.fromkeys(hospitals[item.hospital_id] for item in info
|
|
if hospitals.get(item.hospital_id))),
|
|
salesUserName='、'.join(dict.fromkeys(sales_users[item.sales_user_id] for item in info
|
|
if sales_users.get(item.sales_user_id))),
|
|
)
|
|
if kind == 'cell_storage':
|
|
details['collectionDate'] = '、'.join(dict.fromkeys(
|
|
item.sample_collect_time.strftime('%Y-%m-%d') for item in info if item.sample_collect_time))
|
|
else:
|
|
value = row.donation_date or ''
|
|
details['donationDate'] = f'{value[:4]}-{value[4:6]}-{value[6:]}' if len(value) == 8 else value
|
|
result[row.id] = details
|
|
elif kind == 'pilot_service':
|
|
patients = child_rows(db, tenant, BizPilotPatient, 'pilot_application_id', ids,
|
|
('patient_name', 'phone_no'))
|
|
order_ids = {row.channel_service_order_id for row in rows if row.channel_service_order_id}
|
|
orders = dict(projects.scoped(db, BizChannelServiceOrder, tenant).filter(
|
|
BizChannelServiceOrder.id.in_(order_ids),
|
|
).with_entities(BizChannelServiceOrder.id, BizChannelServiceOrder.order_name).all()) if order_ids else {}
|
|
for row in rows:
|
|
result[row.id] = dict(
|
|
applicationName=row.application_name, channelName=row.channel_name,
|
|
serviceOrderName=orders.get(row.channel_service_order_id),
|
|
patientName=joined_values(patients[row.id], 'patient_name'),
|
|
patientPhone=joined_values(patients[row.id], 'phone_no'),
|
|
)
|
|
else:
|
|
for row in rows:
|
|
result[row.id] = dict(projectTitle=row.project_title, customerName=row.customer_name,
|
|
customerContact=row.customer_contact, customerPhone=row.customer_phone)
|
|
return result
|
|
|
|
|
|
def records(db, tenant, user, biz_type, page=1, pageSize=20, **filters):
|
|
projects.check_permission(db, tenant, user, 'project:assignment:edit')
|
|
return _records(db, tenant, user, biz_type, page, pageSize, **filters)
|
|
|
|
|
|
def statistics_records(db, tenant, user, biz_type, page=1, pageSize=20, **filters):
|
|
projects.check_permission(db, tenant, user, 'project:statistics:list')
|
|
return _records(db, tenant, user, biz_type, page, pageSize, **filters)
|
|
|
|
|
|
def _records(db, tenant, user, biz_type, page, pageSize, **filters):
|
|
if biz_type not in projects.COUNT_TYPES:
|
|
raise HTTPException(400, '业务明细仅支持四类正式业务')
|
|
model, number, status = projects.BUSINESS[biz_type]
|
|
# 继续使用现有租户、主单数据范围、项目/项目组和日期条件。
|
|
query = projects.filtered_query(db, tenant, user, biz_type, {**filters, 'keyword': None})
|
|
keyword = (filters.get('keyword') or '').strip()
|
|
if keyword:
|
|
query = query.filter(keyword_condition(db, tenant, biz_type, keyword))
|
|
status_column = getattr(model, status)
|
|
status_counts = {str(value or ''): count for value, count in query.with_entities(
|
|
status_column, func.count(model.id),
|
|
).group_by(status_column).all()}
|
|
if filters.get('status'):
|
|
query = query.filter(status_column == filters['status'])
|
|
total = query.count()
|
|
rows = query.order_by(model.create_time.desc(), model.id).offset((page - 1) * pageSize).limit(pageSize).all()
|
|
metadata = project_metadata(db, tenant, rows)
|
|
details = business_details(db, tenant, biz_type, rows)
|
|
return dict(list=[dict(
|
|
id=row.id, bizType=biz_type, bizNo=getattr(row, number), status=getattr(row, status),
|
|
createTime=row.create_time, projectId=row.project_id,
|
|
**metadata.get(row.project_id, dict(projectName=None, teamId=None, teamName=None)),
|
|
**details[row.id],
|
|
) for row in rows], total=total, statusCounts=status_counts)
|