from __future__ import annotations from datetime import date, datetime from io import BytesIO import random from typing import Any from openpyxl import Workbook, load_workbook from app.schemas.study_setup_config import StudySetupConfigData def _make_row_id() -> str: return f"{int(datetime.now().timestamp() * 1000)}_{random.randint(1000, 9999)}" def _is_empty(*values: Any) -> bool: return all(v is None or str(v).strip() == "" for v in values) def _sheet_headers(sheet, headers: list[str]) -> None: sheet.append(headers) for idx, title in enumerate(headers, start=1): cell = sheet.cell(row=1, column=idx) cell.value = title def export_setup_config_excel(data: StudySetupConfigData) -> bytes: wb = Workbook() ws_guide = wb.active ws_guide.title = "说明" ws_guide.append(["立项配置Excel模板说明"]) ws_guide.append(["1. 各业务Sheet首行是表头，不要修改列名。"]) ws_guide.append(["2. 可在现有行基础上新增/删除行。"]) ws_guide.append(["3. 日期格式建议：YYYY-MM-DD。"]) ws_guide.append(["4. 导入后系统会自动校验并回填到立项配置草稿。"]) ws_pm = wb.create_sheet("项目里程碑") _sheet_headers(ws_pm, ["ID", "里程碑", "计划日期", "负责人", "状态", "备注"]) for row in data.projectMilestones: ws_pm.append([row.id, row.name, row.planDate, row.owner, row.status, row.remark]) ws_plan = wb.create_sheet("项目入组计划") _sheet_headers(ws_plan, ["计划总入组例数", "计划开始日期", "计划结束日期", "月度目标说明", "分阶段计划"]) p = data.enrollmentPlan ws_plan.append([p.totalTarget, p.startDate, p.endDate, p.monthlyGoalNote, p.stageBreakdown]) ws_sm = wb.create_sheet("中心里程碑") _sheet_headers(ws_sm, ["ID", "里程碑", "计划日期", "负责人", "状态", "备注"]) for row in data.siteMilestones: ws_sm.append([row.id, row.milestone, row.planDate, row.owner, row.status, row.remark]) ws_sep = wb.create_sheet("中心入组计划") _sheet_headers(ws_sep, ["ID", "中心ID", "中心名称", "计划例数", "启动日期", "完成日期", "备注", "分阶段计划"]) for row in data.siteEnrollmentPlans: ws_sep.append([row.id, row.siteId, row.siteName, row.target, row.startDate, row.endDate, row.note, row.stageBreakdown]) ws_ms = wb.create_sheet("监查计划策略") _sheet_headers(ws_ms, ["ID", "监查类型", "策略详情", "监查次数", "更新时间", "是否启用"]) for row in data.monitoringStrategies: ws_ms.append([row.id, row.strategyType, row.detail, row.frequency, row.updatedAt, "是" if row.enabled else "否"]) ws_cc = wb.create_sheet("中心确认") _sheet_headers(ws_cc, ["ID", "中心ID", "中心名称", "确认人", "确认状态", "确认日期", "备注"]) for row in data.centerConfirm: ws_cc.append([row.id, row.siteId, row.siteName, row.confirmer, row.confirmStatus, row.confirmDate, row.note]) bio = BytesIO() wb.save(bio) return bio.getvalue() def _to_int(value: Any, default: int = 0) -> int: if value is None or str(value).strip() == "": return default try: return int(float(str(value))) except (TypeError, ValueError): return default def _to_date_str(value: Any) -> str: if value is None: return "" if isinstance(value, datetime): return value.date().isoformat() if isinstance(value, date): return value.isoformat() return str(value).strip() def _to_datetime_str(value: Any) -> str: if value is None: return "" if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M:%S") return str(value).strip() def _to_bool(value: Any, default: bool = True) -> bool: if value is None: return default v = str(value).strip().lower() if v in {"1", "true", "yes", "y", "是", "启用"}: return True if v in {"0", "false", "no", "n", "否", "禁用"}: return False return default def import_setup_config_excel(file_bytes: bytes) -> StudySetupConfigData: wb = load_workbook(filename=BytesIO(file_bytes), data_only=True) project_milestones = [] ws = wb["项目里程碑"] if "项目里程碑" in wb.sheetnames else None if ws: for row in ws.iter_rows(min_row=2, values_only=True): rid, name, plan_date, owner, status, remark = row[:6] if _is_empty(rid, name, plan_date, owner, status, remark): continue project_milestones.append( { "id": str(rid).strip() if rid else _make_row_id(), "name": str(name or "").strip(), "planDate": _to_date_str(plan_date), "owner": str(owner or "").strip(), "status": str(status or "未开始").strip() or "未开始", "remark": str(remark or "").strip(), } ) enrollment_plan = { "totalTarget": 0, "startDate": "", "endDate": "", "monthlyGoalNote": "", "stageBreakdown": "", } ws = wb["项目入组计划"] if "项目入组计划" in wb.sheetnames else None if ws: first = next(ws.iter_rows(min_row=2, values_only=True), None) if first: enrollment_plan = { "totalTarget": _to_int(first[0], 0), "startDate": _to_date_str(first[1]), "endDate": _to_date_str(first[2]), "monthlyGoalNote": str(first[3] or "").strip(), "stageBreakdown": str(first[4] or "").strip(), } site_milestones = [] ws = wb["中心里程碑"] if "中心里程碑" in wb.sheetnames else None if ws: for row in ws.iter_rows(min_row=2, values_only=True): raw_status = None if len(row) >= 8: rid, site_id, site_name, milestone, plan_date, owner, raw_status, remark = row[:8] _ = (site_id, site_name) elif len(row) == 7: # Backward compatibility: old template has no status column and includes center columns. rid, site_id, site_name, milestone, plan_date, owner, remark = row[:7] _ = (site_id, site_name) else: rid, milestone, plan_date, owner, raw_status, remark = row[:6] if _is_empty(rid, milestone, plan_date, owner, raw_status, remark): continue status = str(raw_status or "未开始").strip() or "未开始" site_milestones.append( { "id": str(rid).strip() if rid else _make_row_id(), "milestone": str(milestone or "").strip(), "planDate": _to_date_str(plan_date), "owner": str(owner or "").strip(), "status": status, "remark": str(remark or "").strip(), } ) site_enrollment_plans = [] ws = wb["中心入组计划"] if "中心入组计划" in wb.sheetnames else None if ws: for row in ws.iter_rows(min_row=2, values_only=True): rid, site_id, site_name, target, start_date, end_date, note, stage_breakdown = (list(row[:8]) + [None] * 8)[:8] if _is_empty(rid, site_id, site_name, target, start_date, end_date, note, stage_breakdown): continue site_enrollment_plans.append( { "id": str(rid).strip() if rid else _make_row_id(), "siteId": str(site_id or "").strip(), "siteName": str(site_name or "").strip(), "target": _to_int(target, 0), "startDate": _to_date_str(start_date), "endDate": _to_date_str(end_date), "note": str(note or "").strip(), "stageBreakdown": str(stage_breakdown or "").strip(), } ) monitoring_strategies = [] ws = wb["监查计划策略"] if "监查计划策略" in wb.sheetnames else None if ws: for row in ws.iter_rows(min_row=2, values_only=True): rid, strategy_type, detail, frequency, updated_at, enabled = row[:6] if _is_empty(rid, strategy_type, detail, frequency, updated_at, enabled): continue monitoring_strategies.append( { "id": str(rid).strip() if rid else _make_row_id(), "strategyType": str(strategy_type or "").strip(), "detail": str(detail or "").strip(), "frequency": str(frequency or "").strip(), "updatedAt": _to_datetime_str(updated_at) or datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "enabled": _to_bool(enabled, True), } ) center_confirm = [] ws = wb["中心确认"] if "中心确认" in wb.sheetnames else None if ws: for row in ws.iter_rows(min_row=2, values_only=True): rid, site_id, site_name, confirmer, confirm_status, confirm_date, note = row[:7] if _is_empty(rid, site_id, site_name, confirmer, confirm_status, confirm_date, note): continue center_confirm.append( { "id": str(rid).strip() if rid else _make_row_id(), "siteId": str(site_id or "").strip(), "siteName": str(site_name or "").strip(), "confirmer": str(confirmer or "").strip(), "confirmStatus": str(confirm_status or "待确认").strip() or "待确认", "confirmDate": _to_date_str(confirm_date), "note": str(note or "").strip(), } ) return StudySetupConfigData.model_validate( { "projectMilestones": project_milestones, "enrollmentPlan": enrollment_plan, "siteMilestones": site_milestones, "siteEnrollmentPlans": site_enrollment_plans, "monitoringStrategies": monitoring_strategies, "centerConfirm": center_confirm, } )