255 lines
8.7 KiB
Python
255 lines
8.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""按生成质量投诉模拟数据.md生成质量投诉模拟数据。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import random
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from openpyxl.utils import get_column_letter
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
TEMPLATE = ROOT / "质量投诉数据-表头.xlsx"
|
|
|
|
PRODUCTS = [
|
|
{
|
|
"division": "输液治疗",
|
|
"bu": "IV Therapy BU",
|
|
"name": "一次性使用静脉留置针",
|
|
"models": ["20G", "22G", "24G", "Introcan Safety 20G"],
|
|
"faults": [
|
|
("渗漏", "使用中发现连接处渗液，影响持续输注"),
|
|
("连接不牢", "与延长管连接后松动，需重复连接"),
|
|
("流速异常", "滴速不稳定，出现间断性停止"),
|
|
("包装破损", "开包前发现无菌包装破损"),
|
|
],
|
|
"cases_max": 3,
|
|
},
|
|
{
|
|
"division": "输液治疗",
|
|
"bu": "IV Therapy BU",
|
|
"name": "一次性使用输液器",
|
|
"models": ["IS-5.0", "IS-7.0", "PF-5.0"],
|
|
"faults": [
|
|
("堵塞", "输液过程中阻力升高，液体无法正常滴注"),
|
|
("渗漏", "滴斗下方接口渗漏，需更换器械"),
|
|
("标签不清", "批号标签印刷不清，追溯困难"),
|
|
("流速异常", "调节轮无法稳定控制滴速"),
|
|
],
|
|
"cases_max": 4,
|
|
},
|
|
{
|
|
"division": "透析",
|
|
"bu": "Renal Care BU",
|
|
"name": "血液透析器",
|
|
"models": ["1.3m²", "1.4m²", "HD-180"],
|
|
"faults": [
|
|
("连接不牢", "血路连接后出现轻度渗血风险"),
|
|
("断裂", "外壳边缘出现细微裂纹，已停止使用"),
|
|
("流速异常", "跨膜压波动导致透析流程中断"),
|
|
("包装破损", "透析器外包装密封不完整"),
|
|
],
|
|
"cases_max": 2,
|
|
},
|
|
{
|
|
"division": "透析",
|
|
"bu": "Renal Care BU",
|
|
"name": "血液透析管路",
|
|
"models": ["AV-SET-A", "AV-SET-P"],
|
|
"faults": [
|
|
("渗漏", "动脉端管路连接位出现渗液"),
|
|
("堵塞", "泵管段阻力偏高触发报警"),
|
|
("连接不牢", "静脉端接头锁定后仍可松脱"),
|
|
],
|
|
"cases_max": 2,
|
|
},
|
|
{
|
|
"division": "外科",
|
|
"bu": "Surgical BU",
|
|
"name": "可吸收性外科缝线",
|
|
"models": ["3-0", "4-0", "5-0"],
|
|
"faults": [
|
|
("断裂", "缝合过程中线体提前断裂"),
|
|
("标签不清", "规格标签与实际颜色识别困难"),
|
|
("包装破损", "单支包装封口松开，未使用"),
|
|
],
|
|
"cases_max": 2,
|
|
},
|
|
]
|
|
|
|
HOSPITALS = [
|
|
"上海市第一人民医院",
|
|
"浙江大学医学院附属第二医院",
|
|
"四川大学华西医院",
|
|
"广东省人民医院",
|
|
"中南大学湘雅医院",
|
|
"山东大学齐鲁医院",
|
|
"西安交通大学第一附属医院",
|
|
"南京鼓楼医院",
|
|
"福建省立医院",
|
|
"郑州大学第一附属医院",
|
|
]
|
|
|
|
CONTACTS = ["张医生", "李护士长", "王老师", "陈医生", "赵工", "刘老师", "周护士"]
|
|
REPORTERS = ["设备科-赵工", "护理部-陈老师", "质控办-王老师", "临床工程-李工"]
|
|
MANUFACTURERS = [
|
|
"贝朗医疗（上海）国际贸易有限公司",
|
|
"贝朗医疗（苏州）有限公司",
|
|
"贝朗爱敦（上海）医疗管理有限公司",
|
|
]
|
|
AE_NEG_REASONS = ["未造成患者伤害", "仅质量缺陷，无临床后果", "使用前发现缺陷，未接触患者"]
|
|
CONCLUSIONS = ["产品缺陷成立", "操作不当", "运输损伤", "未复现", "资料不足"]
|
|
PAYMENTS = ["无赔付", "换货", "折让", "退款", "其他协商处理"]
|
|
STATUS_POOL = ["新建", "调查中", "待补充", "已关闭"]
|
|
|
|
|
|
def rand_reg_no(rng: random.Random) -> str:
|
|
return f"国械注进20{rng.randint(15,26)}{rng.randint(100000,999999)}"
|
|
|
|
|
|
def rand_batch(rng: random.Random) -> str:
|
|
return f"{rng.choice('ABCDEFGH')}{rng.randint(1000000,9999999)}"
|
|
|
|
|
|
def rand_phone(rng: random.Random) -> str:
|
|
return f"1{rng.choice('3456789')}{rng.randint(100,999)}****{rng.randint(1000,9999)}"
|
|
|
|
|
|
def rand_serial(rng: random.Random, use_na: bool) -> str:
|
|
if use_na:
|
|
return "N/A"
|
|
return f"SN{rng.randint(10**9,10**10-1)}"
|
|
|
|
|
|
def choose_resolution(rng: random.Random, conclusion: str) -> tuple[str, str]:
|
|
if conclusion == "产品缺陷成立":
|
|
return "Replace", "更换同批次产品并加强到货检验"
|
|
if conclusion == "操作不当":
|
|
return "Training reinforced", "复测未见异常，建议规范操作培训"
|
|
if conclusion == "运输损伤":
|
|
return "Logistics improved", "判定为运输环节损伤，已优化包装与交付流程"
|
|
if conclusion == "未复现":
|
|
return "No defect found", "复测未见异常，建议持续观察后续批次"
|
|
return "Need more info", "现有证据不足，待补充样品及记录"
|
|
|
|
|
|
def generate_rows(n: int, seed: int) -> list[list]:
|
|
rng = random.Random(seed)
|
|
rows: list[list] = []
|
|
today = date.today()
|
|
start = today - timedelta(days=980)
|
|
|
|
for i in range(1, n + 1):
|
|
p = rng.choice(PRODUCTS)
|
|
fault, detail = rng.choice(p["faults"])
|
|
hospital = rng.choice(HOSPITALS)
|
|
c3_date = start + timedelta(days=rng.randint(0, (today - start).days))
|
|
c3_month = c3_date.strftime("%Y-%m")
|
|
|
|
is_ae = "是" if rng.random() < 0.28 else "否"
|
|
ae_no_reason = "" if is_ae == "是" else rng.choice(AE_NEG_REASONS)
|
|
|
|
status = rng.choices(STATUS_POOL, weights=[0.15, 0.25, 0.1, 0.5], k=1)[0]
|
|
report_done = None
|
|
closed_date = None
|
|
if status in ("调查中", "已关闭", "待补充"):
|
|
report_done = c3_date + timedelta(days=rng.randint(3, 45))
|
|
if status == "已关闭":
|
|
base = report_done if report_done else c3_date
|
|
closed_date = base + timedelta(days=rng.randint(1, 30))
|
|
|
|
bad_qty = rng.randint(1, 20)
|
|
sample_return = "是" if rng.random() < 0.78 else "否"
|
|
if sample_return == "否":
|
|
qa_qty = 0 if rng.random() < 0.95 else rng.randint(0, bad_qty)
|
|
else:
|
|
qa_qty = rng.randint(1, bad_qty)
|
|
factory_qty = rng.randint(0, qa_qty)
|
|
case_count = 1 if rng.random() < 0.85 else rng.randint(2, p["cases_max"])
|
|
|
|
conclusion = rng.choice(CONCLUSIONS)
|
|
payment = rng.choice(PAYMENTS)
|
|
opinion_en, opinion_cn = choose_resolution(rng, conclusion)
|
|
|
|
# make payment more coherent
|
|
if conclusion in ("未复现", "资料不足", "操作不当"):
|
|
payment = rng.choices(["无赔付", "其他协商处理", "折让"], weights=[0.7, 0.2, 0.1], k=1)[0]
|
|
elif conclusion == "产品缺陷成立":
|
|
payment = rng.choices(["换货", "退款", "折让", "无赔付"], weights=[0.5, 0.2, 0.2, 0.1], k=1)[0]
|
|
|
|
row = [
|
|
f"C3-{today:%Y}-{i:06d}",
|
|
rng.choice(p["models"]),
|
|
rand_batch(rng),
|
|
rand_serial(rng, use_na=("输液" in p["division"] and rng.random() < 0.6)),
|
|
rng.choice(MANUFACTURERS),
|
|
rand_reg_no(rng),
|
|
p["name"],
|
|
hospital,
|
|
rng.choice(CONTACTS),
|
|
rand_phone(rng),
|
|
fault,
|
|
detail,
|
|
rng.choice(REPORTERS),
|
|
p["bu"],
|
|
c3_date,
|
|
c3_month,
|
|
is_ae,
|
|
ae_no_reason,
|
|
bad_qty,
|
|
qa_qty,
|
|
factory_qty,
|
|
report_done,
|
|
opinion_en,
|
|
opinion_cn,
|
|
conclusion,
|
|
payment,
|
|
closed_date,
|
|
status,
|
|
sample_return,
|
|
p["division"],
|
|
case_count,
|
|
]
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def template_headers() -> list[str]:
|
|
wb = load_workbook(TEMPLATE, read_only=True, data_only=True)
|
|
ws = wb.active
|
|
headers = [ws.cell(1, j).value for j in range(1, ws.max_column + 1)]
|
|
wb.close()
|
|
return headers
|
|
|
|
|
|
def main() -> None:
|
|
headers = template_headers()
|
|
today = date.today()
|
|
out_path = ROOT / f"质量投诉数据-模拟1000条-{today:%Y%m%d}.xlsx"
|
|
rows = generate_rows(1000, int(today.strftime("%Y%m%d")))
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "complaint form"
|
|
ws.append(headers)
|
|
for r in rows:
|
|
ws.append(r)
|
|
|
|
# date columns
|
|
for r in range(2, 1002):
|
|
for col in ("O", "V", "AA"):
|
|
if ws[f"{col}{r}"].value:
|
|
ws[f"{col}{r}"].number_format = "YYYY-MM-DD"
|
|
|
|
for i in range(1, len(headers) + 1):
|
|
ws.column_dimensions[get_column_letter(i)].width = 16
|
|
|
|
wb.save(out_path)
|
|
print(f"Written: {out_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|