diff --git a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/TherapyRecuperationModule.java b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/TherapyRecuperationModule.java index 22ee3b9..814c87e 100644 --- a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/TherapyRecuperationModule.java +++ b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/TherapyRecuperationModule.java @@ -49,7 +49,6 @@ public class TherapyRecuperationModule { private SysUserService sysUserService; public void init() { - renameLegacyTables(); Class[] tables = {TheRapyRecuperationBaseManagement.class, TheRapyRecuperationCluster.class, TheRapyRecuperationClusterMember.class, TheRapyRecuperationConfig.class, TheRapyRecuperationEnroll.class, TheRapyRecuperationEnrollBed.class, @@ -58,7 +57,6 @@ public class TherapyRecuperationModule { TheRapyRecuperationLot.class, TheRapyRecuperationTravelAgency.class, TherapyAudit.class, TherapyAuditState.class}; for (Class table : tables) dao.create(table, false); - widenLineImageAddress(); createView("inclusive_therapy_v3_user", "SELECT u.*, u.arrivalAtSchoolDate AS schoolTime FROM vw_user u"); createView("inclusive_therapy_v3_union", "SELECT u.*, u.name AS unionname FROM sys_union u"); createView("inclusive_therapy_v3_unit", "SELECT u.* FROM sys_unit u"); @@ -116,7 +114,6 @@ public class TherapyRecuperationModule { } private void seedMenus() { - migrateLegacyMenuIdentifiers(); Sys_menu existing = dao.fetch(Sys_menu.class, Cnd.where("href", "like", "/platform/recuperation%").and("platform", "=", "pc")); Sys_module module = existing == null ? null : dao.fetch(Sys_module.class, existing.getModuleId()); if (module == null) module = dao.fetch(Sys_module.class, Cnd.where("name", "=", "职工福利").and("platform", "=", "pc")); diff --git a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/controller/TherapyCommonController.java b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/controller/TherapyCommonController.java index 218840a..cc97265 100644 --- a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/controller/TherapyCommonController.java +++ b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/controller/TherapyCommonController.java @@ -51,7 +51,8 @@ public class TherapyCommonController { @At({"/unions", "/getUnion"}) @SaCheckLogin public Result unions(Boolean isPermission) { - Cnd cnd = Cnd.NEW().asc("unionname"); + Cnd cnd = Cnd.NEW(); + cnd.asc("unionname"); if (Boolean.TRUE.equals(isPermission) && !TherapyContext.hasAnyRoles("sysadmin,A06")) { cnd.and("id", "=", TherapyContext.getUnionId()); } diff --git a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/service/impl/TheRapyRecuperationEnrollServiceImpl.java b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/service/impl/TheRapyRecuperationEnrollServiceImpl.java index 51a8a6e..4c80b55 100644 --- a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/service/impl/TheRapyRecuperationEnrollServiceImpl.java +++ b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/service/impl/TheRapyRecuperationEnrollServiceImpl.java @@ -1790,7 +1790,7 @@ public class TheRapyRecuperationEnrollServiceImpl extends TherapyServiceImpl 0) { str = "省内线路"; @@ -1855,8 +1855,8 @@ public class TheRapyRecuperationEnrollServiceImpl extends TherapyServiceImpl=", com.budwk.app.base.utils.DateUtil.thisYear() - outsideNumber); - lineCnd.and("YEAR ( signingUptime )", "<=", com.budwk.app.base.utils.DateUtil.thisYear() - 1); + lineCnd.and("YEAR ( signingUptime )", ">=", DateUtil.thisYear() - outsideNumber); + lineCnd.and("YEAR ( signingUptime )", "<=", DateUtil.thisYear() - 1); lineCnd.and("regionalNature", "=", TheRapyRecuperationType.provinceOutLine.getValue()); lineJoinSql.setCondition(lineCnd); diff --git a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/support/TherapyContext.java b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/support/TherapyContext.java index cca6446..97485fd 100644 --- a/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/support/TherapyContext.java +++ b/src/main/java/com/budwk/app/zhgh/staffbenefit/therapyRecuperation/support/TherapyContext.java @@ -1,6 +1,7 @@ package com.budwk.app.zhgh.staffbenefit.therapyRecuperation.support; import cn.dev33.satoken.stp.StpUtil; +import com.budwk.app.sys.services.SysUserService; import com.budwk.app.web.commons.auth.utils.SecurityUtil; import org.nutz.dao.Dao; import org.nutz.ioc.loader.annotation.IocBean; @@ -20,6 +21,10 @@ public class TherapyContext { return getUserId(); } + public static String getPlatformLoginname() { + return SecurityUtil.getUserLoginname(); + } + public static String getUnionId() { String unionId = SecurityUtil.getUnionId(); if (Strings.isNotBlank(unionId)) return unionId; @@ -57,7 +62,8 @@ public class TherapyContext { } public static String getRoleCodesJson() { - java.util.List roles = new java.util.ArrayList<>(StpUtil.getRoleList()); + SysUserService userService = Mvcs.getIoc().get(SysUserService.class); + java.util.List roles = new java.util.ArrayList<>(userService.getRoleCodeList(userService.fetch(getUserId()))); for (String role : Arrays.asList("sysadmin", "A06", "H03", "H04", "H06", "SchoolUnionMemberAdmin")) { if (hasRole(role) && !roles.contains(role)) roles.add(role); } diff --git a/src/main/resources/application-dev.yaml b/src/main/resources/application-dev.yaml index 5f69385..ea7aa76 100644 --- a/src/main/resources/application-dev.yaml +++ b/src/main/resources/application-dev.yaml @@ -42,8 +42,8 @@ sdk: client-call-back-url: /platform/sso/login validation-type: cas minio: - accessKey: o1MwuzixNqAXh5zqEfMH - secretKey: RVnZum91JmGO79qtzWy2gKMBsF6OxAaibGrr7ets + accessKey: minioadmin + secretKey: Minio@123#admin endPoint: http://localhost:9000 bucket: zjnu diff --git a/src/main/resources/db/v4_from_v3/export_inclusive_therapy_v3.py b/src/main/resources/db/v4_from_v3/export_inclusive_therapy_v3.py index 87d2864..44c583f 100644 --- a/src/main/resources/db/v4_from_v3/export_inclusive_therapy_v3.py +++ b/src/main/resources/db/v4_from_v3/export_inclusive_therapy_v3.py @@ -8,6 +8,7 @@ """ import argparse +from datetime import datetime import getpass import hashlib import json @@ -17,6 +18,7 @@ import re import subprocess import sys import tempfile +import time from urllib.parse import unquote, urlsplit import zipfile @@ -31,6 +33,11 @@ SUFFIXES = ( AUDIT_FIELDS = ("selfUnionAuditId", "joinLineUnionAuditId", "schoolUnionAuditId") +def log(message, error=False): + print("[%s] %s" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), message), + file=sys.stderr if error else sys.stdout, flush=True) + + def identifier(value): if not re.fullmatch(r"[A-Za-z0-9_]+", value): raise ValueError("数据库标识符只允许英文字母、数字、下划线") @@ -148,6 +155,8 @@ def main(): output = args.output.resolve() if output.exists(): raise ValueError("导出文件已存在，不覆盖：" + str(output)) + log("开始导出普惠疗休养V3；源数据库：" + args.database) + log("[1/4] 连接数据库并检查源表结构") client = MySQL(args) tables = {suffix: "the_rapy_recuperation_" + suffix for suffix in SUFFIXES} schemas = client.schema(list(tables.values()) + [args.audit_table, "activity_user_scope", @@ -192,24 +201,40 @@ def main(): archive = zipfile.ZipFile(output, "x", compression=zipfile.ZIP_DEFLATED, allowZip64=True) finished = False try: + log("[2/4] 从一致性快照读取数据并写入导出包") + current_label = None + total_records = 0 with archive.open("data.jsonl", "w", force_zip64=True) as data: for line in client.lines(sql): record = json.loads(line) record["row"] = decode_row(record) + if record["table"] != current_label: + if current_label is not None: + log("数据类别 %s 已导出 %s 条" % (current_label, counts[current_label])) + current_label = record["table"] + log("正在导出数据类别：" + current_label) counts[record["table"]] += 1 if record["table"] == "_file": file_rows.append(record["row"]) encoded = (json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8") digest.update(encoded) data.write(encoded) + total_records += 1 + if total_records % 1000 == 0: + log("已写入 %s 条记录" % total_records) + if current_label is not None: + log("数据类别 %s 已导出 %s 条" % (current_label, counts[current_label])) + log("数据库读取完成，共 %s 条记录" % total_records) + log("[3/4] 打包实际附件并计算 SHA-256，共 %s 个" % len(file_rows)) assets = {} if file_rows and args.files_root is None: raise ValueError("数据包含附件，必须提供 --files-root，不能只导出文件地址") storage = args.files_root.resolve() if args.files_root else None - for row in file_rows: + for index, row in enumerate(file_rows, 1): old_id, old_path = get(row, "id"), get(row, "filepath") if not old_id or not old_path: raise ValueError("关联附件记录缺少 ID 或存储路径") + log("附件 [%s/%s] 开始打包，文件ID：%s" % (index, len(file_rows), old_id)) relative = unquote(urlsplit(old_path.replace("\\", "/")).path).lstrip("/") source = (storage / relative).resolve() if not source.is_relative_to(storage) or not source.is_file(): @@ -223,6 +248,10 @@ def main(): size += len(chunk) writer.write(chunk) assets[old_id] = {"member": member, "sha256": file_hash.hexdigest(), "size": size} + log("附件 [%s/%s] 打包完成（%s 字节），SHA-256 已记录" % (index, len(file_rows), size)) + if not file_rows: + log("没有关联附件，跳过附件打包") + log("[4/4] 写入清单并关闭导出包") manifest = {"format": FORMAT, "source_database": args.database, "source_prefix": "the_rapy_recuperation_", "target_prefix": PREFIX, "counts": counts, "data_sha256": digest.hexdigest(), "files": assets} archive.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8")) @@ -231,14 +260,20 @@ def main(): archive.close() if not finished: output.unlink() - print("导出完成：" + str(output)) - print("业务记录：%s；审核记录：%s；附件：%s" % - (sum(counts.get(name, 0) for name in SUFFIXES), counts.get("_audit", 0), len(file_rows))) + log("导出完成：" + str(output)) + log("业务记录：%s；审核记录：%s；附件：%s" % + (sum(counts.get(name, 0) for name in SUFFIXES), counts.get("_audit", 0), len(file_rows))) if __name__ == "__main__": + started = time.monotonic() try: main() + log("【成功结束】导出成功，总耗时 %.1f 秒，退出码 0" % (time.monotonic() - started)) + except KeyboardInterrupt: + log("【中断结束】用户中断导出，总耗时 %.1f 秒，退出码 130" % (time.monotonic() - started), error=True) + sys.exit(130) except (OSError, ValueError, RuntimeError, KeyError) as error: - print("导出失败：" + str(error), file=sys.stderr) + log("导出失败：" + str(error), error=True) + log("【失败结束】导出未成功，总耗时 %.1f 秒，退出码 1" % (time.monotonic() - started), error=True) sys.exit(1) diff --git a/src/main/resources/db/v4_from_v3/import_inclusive_therapy_v3.py b/src/main/resources/db/v4_from_v3/import_inclusive_therapy_v3.py index c55f7e3..9b2e797 100644 --- a/src/main/resources/db/v4_from_v3/import_inclusive_therapy_v3.py +++ b/src/main/resources/db/v4_from_v3/import_inclusive_therapy_v3.py @@ -2,23 +2,34 @@ 两个 Python 脚本应放在同一目录；使用 Python 3.9+ 标准库及已有 mysql 客户端。 参数：--database v4目标库 --input 导出.zip --login-path 连接名。 -有附件时 --target-files-dir 必须是 v4 服务器可直接访问的本地/共享存储目录。 +有附件时通过 MinIO S3 API 上传，不复制到本地目录或 MinIO 底层存储目录。 +--minio-endpoint、--minio-bucket、--minio-access-key 分别对应 v4 的 MinIO 端点、桶和访问密钥。 +也可使用 MINIO_ENDPOINT、MINIO_BUCKET、MINIO_ACCESS_KEY 环境变量；MINIO_SECRET_KEY 未设置时交互输入。 +自定义区域使用 --minio-region / MINIO_REGION；临时凭证可设置 MINIO_SESSION_TOKEN。 先使用本目录初始化 SQL 直接创建 inclusive_therapy_v3_* 表及菜单。 仅支持首次导入：目标子模块必须为空，或仅包含未使用的自动生成默认配置。 可用 --mapping 提供 JSON：{"user":{"旧ID":"新ID"},"unit":{},"union":{}}。 不覆盖 v4 原有疗休养、人员、组织或附件；导入期间不要向目标库写入业务数据。 -失败时数据库事务回滚；已复制的附件保留在独立目录，避免提交状态不明时误删文件。 +附件逐个上传并回读校验后才写入数据库；失败时保留独立对象前缀，避免提交状态不明时误删附件。 """ import argparse from collections import defaultdict +from datetime import datetime, timezone +import getpass import hashlib +import hmac +import http.client import json +import mimetypes +import os from pathlib import Path import re import sys +import tempfile +import time import uuid -from urllib.parse import parse_qs, unquote, urlsplit +from urllib.parse import parse_qs, quote, unquote, urlsplit import zipfile from export_inclusive_therapy_v3 import ( @@ -27,6 +38,11 @@ from export_inclusive_therapy_v3 import ( ) +def log(message, error=False): + print("[%s] %s" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), message), + file=sys.stderr if error else sys.stdout, flush=True) + + def rows(client, table, columns): from export_inclusive_therapy_v3 import row_query, decode_row return [decode_row(json.loads(line)) for line in client.lines(row_query(table, table, columns))] @@ -142,18 +158,129 @@ def default_config(row): return True +class MinIO: + """使用标准库签名并调用 S3 API，不依赖额外 Python 包。 + + 签名格式：https://docs.aws.amazon.com/AmazonS3/latest/developerguide/sig-v4-header-based-auth.html + """ + + def __init__(self, args): + if not args.minio_endpoint or not args.minio_bucket or not args.minio_access_key: + raise ValueError("有附件时必须提供与 v4 一致的 MinIO 端点、桶和访问密钥") + self.endpoint = urlsplit(args.minio_endpoint) + if (self.endpoint.scheme not in ("http", "https") or not self.endpoint.hostname + or self.endpoint.username or self.endpoint.password or self.endpoint.query or self.endpoint.fragment): + raise ValueError("MinIO 端点必须是 http/https S3 API 地址，不能包含账号、查询参数或片段") + self.port = self.endpoint.port + self.bucket = args.minio_bucket + self.access_key = args.minio_access_key + self.region = args.minio_region + self.session_token = os.environ.get("MINIO_SESSION_TOKEN") + self.secret_key = os.environ.get("MINIO_SECRET_KEY") or getpass.getpass("MinIO secret key: ") + if not self.secret_key or not self.region or "/" in self.bucket: + raise ValueError("MinIO 桶、区域或密钥配置不正确") + for value in (self.bucket, self.access_key, self.region, self.session_token or ""): + if re.search(r"[\x00-\x20\x7f]", value): + raise ValueError("MinIO 连接参数包含不合法的空白或控制字符") + + def request(self, method, key=None, body=None, size=0, digest=None, content_type=None): + uri = self.endpoint.path.rstrip("/") + "/" + quote(self.bucket, safe="-_.~") + if key is not None: + uri += "/" + quote(key, safe="/-_.~") + # 对 URI 只编码对象名，不改变对象路径；不自动跟随重定向转发凭证。 + now = datetime.now(timezone.utc) + timestamp, date = now.strftime("%Y%m%dT%H%M%SZ"), now.strftime("%Y%m%d") + payload_hash = digest if digest is not None else hashlib.sha256(b"").hexdigest() + headers = {"host": self.endpoint.netloc, "x-amz-date": timestamp, + "x-amz-content-sha256": payload_hash} + if self.session_token: + headers["x-amz-security-token"] = self.session_token + if method == "PUT": + headers.update({"content-length": str(size), "content-type": content_type, + "x-amz-meta-sha256": payload_hash, "if-none-match": "*"}) + signed_headers = ";".join(sorted(headers)) + canonical_headers = "".join(name + ":" + headers[name].strip() + "\n" for name in sorted(headers)) + canonical_request = "\n".join((method, uri, "", canonical_headers, signed_headers, payload_hash)) + scope = date + "/" + self.region + "/s3/aws4_request" + string_to_sign = "\n".join(("AWS4-HMAC-SHA256", timestamp, scope, + hashlib.sha256(canonical_request.encode("utf-8")).hexdigest())) + signing_key = ("AWS4" + self.secret_key).encode("utf-8") + for value in (date, self.region, "s3", "aws4_request"): + signing_key = hmac.new(signing_key, value.encode("utf-8"), hashlib.sha256).digest() + signature = hmac.new(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest() + headers["authorization"] = ("AWS4-HMAC-SHA256 Credential=" + self.access_key + "/" + scope + + ", SignedHeaders=" + signed_headers + ", Signature=" + signature) + connection_type = http.client.HTTPSConnection if self.endpoint.scheme == "https" else http.client.HTTPConnection + connection = connection_type(self.endpoint.hostname, self.port, timeout=120) + try: + connection.request(method, uri, body=body, headers=headers) + response = connection.getresponse() + if not 200 <= response.status < 300: + status = response.status + response.close() + raise RuntimeError("MinIO %s 请求失败（HTTP %s），数据库尚未写入" % (method, status)) + return connection, response + except (OSError, http.client.HTTPException) as error: + connection.close() + raise RuntimeError("MinIO 网络请求失败，数据库尚未写入") from error + except BaseException: + connection.close() + raise + + def check_bucket(self): + connection, response = self.request("HEAD") + try: + response.read() + finally: + response.close() + connection.close() + + def upload_and_verify(self, key, body, size, digest, content_type): + connection, response = self.request("PUT", key, body, size, digest, content_type) + try: + response.read() + except (OSError, http.client.HTTPException) as error: + raise RuntimeError("MinIO 上传响应读取失败，数据库尚未写入") from error + finally: + response.close() + connection.close() + connection, response = self.request("GET", key) + try: + actual_hash, actual_size = hashlib.sha256(), 0 + for chunk in iter(lambda: response.read(1024 * 1024), b""): + actual_size += len(chunk) + actual_hash.update(chunk) + if actual_size > size: + raise RuntimeError("MinIO 回读对象的文件大小与导出附件不一致") + if actual_size != size or actual_hash.hexdigest() != digest: + raise RuntimeError("MinIO 回读对象的大小或 SHA-256 与导出附件不一致") + except (OSError, http.client.HTTPException) as error: + raise RuntimeError("MinIO 对象回读失败，数据库尚未写入") from error + finally: + response.close() + connection.close() + + def main(): parser = argparse.ArgumentParser(description=__doc__) connection_args(parser) parser.add_argument("--input", type=Path, required=True) - parser.add_argument("--target-files-dir", type=Path) + parser.add_argument("--minio-endpoint", default=os.environ.get("MINIO_ENDPOINT")) + parser.add_argument("--minio-bucket", default=os.environ.get("MINIO_BUCKET")) + parser.add_argument("--minio-access-key", default=os.environ.get("MINIO_ACCESS_KEY")) + parser.add_argument("--minio-region", default=os.environ.get("MINIO_REGION", "us-east-1")) parser.add_argument("--mapping", type=Path) args = parser.parse_args() + log("开始导入普惠疗休养V3；目标数据库：" + args.database) + log("[1/5] 读取导出包并校验数据：" + str(args.input)) overrides = json.loads(args.mapping.read_text(encoding="utf-8-sig")) if args.mapping else {} - copied_directory = None + uploaded_location = None with zipfile.ZipFile(args.input) as archive: manifest, records = load_bundle(archive) check_relations(records) + log("导出包校验通过：业务记录 %s 条，审核记录 %s 条，附件 %s 个" % + (sum(len(records[name]) for name in SUFFIXES), len(records["_audit"]), len(records["_file"]))) + log("[2/5] 连接数据库，检查表结构、目标数据和人员组织映射") client = MySQL(args) tables = {suffix: PREFIX + suffix for suffix in SUFFIXES} tables["_audit"] = PREFIX + "audit" @@ -178,6 +305,14 @@ def main(): initial = rows(client, tables["config"], schemas[tables["config"]]) if len(initial) > 1 or initial and not default_config(initial[0]): raise ValueError("目标配置已有业务设置，不覆盖；仅允许空的新子模块首次导入") + for label, table in tables.items(): + if label != "config": + count = client.objects("SELECT JSON_OBJECT('n',COUNT(*)) FROM " + identifier(table) + ";")[0]["n"] + if count: + raise ValueError("目标子模块已有业务数据，不覆盖：" + table) + log("[3/5] 转换业务数据和附件引用") + minio = MinIO(args) if records["_file"] else None + object_prefix = "inclusive_therapy_v3/" + uuid.uuid4().hex file_info = {} paths = {} for row in records["_file"]: @@ -185,9 +320,20 @@ def main(): asset = manifest["files"].get(old_id) if not asset or asset["member"] not in archive.namelist(): raise ValueError("附件内容缺失，ID：" + old_id) + if type(asset.get("size")) is not int or asset["size"] < 0 or not re.fullmatch(r"[0-9a-f]{64}", asset.get("sha256", "")): + raise ValueError("附件大小或 SHA-256 信息不正确，ID：" + old_id) new_id = uuid.uuid4().hex + name = get(row, "filename") or old_id + suffix = Path(name).suffix.lower().lstrip(".") + if not re.fullmatch(r"[a-z0-9]{1,12}", suffix): + suffix = "bin" + object_name = new_id + "." + suffix file_info[old_id] = {"id": new_id, "name": get(row, "filename") or old_id, - "engine": "LOCAL", "bucket": "", "downloadPath": "/platform/sys/file/download?id=" + new_id} + "engine": "MINIO", "bucket": minio.bucket, + "storagePath": object_prefix + "/" + object_name, "objName": object_name, + "suffix": suffix, "sizeKb": str(round(asset["size"] / 1024, 2)), + "sizeInfo": str(asset["size"]) + " B", "delFlag": "0", + "downloadPath": "/platform/sys/file/download?id=" + new_id} path = get(row, "filepath") if path in paths and paths[path] != old_id: raise ValueError("源附件路径对应多个文件 ID，请先消除歧义") @@ -276,30 +422,35 @@ def main(): version = get(records["_version"][0], "configValue") if records["_version"] else "zjnu" if version not in ("zjnu", "zjxu", "zjiet"): raise ValueError("无法识别 v3 的疗休养规则版本") + log("[4/5] 上传附件到 MinIO 并回读校验，共 %s 个" % len(file_info)) if file_info: - if args.target_files_dir is None: - raise ValueError("有附件时必须提供 --target-files-dir，不能只迁移数据库地址") - storage = args.target_files_dir.resolve() - storage.mkdir(parents=True, exist_ok=True) - copied_directory = storage / ("inclusive_therapy_v3_" + uuid.uuid4().hex) - copied_directory.mkdir() - for old_id, info in file_info.items(): - asset = manifest["files"][old_id] - suffix = Path(info["name"]).suffix.lower().lstrip(".") - if not re.fullmatch(r"[a-z0-9]{1,12}", suffix): - suffix = "bin" - path = copied_directory / (info["id"] + "." + suffix) - digest, size = hashlib.sha256(), 0 - with archive.open(asset["member"]) as source, path.open("xb") as destination: - for chunk in iter(lambda: source.read(1024 * 1024), b""): - digest.update(chunk) - size += len(chunk) - destination.write(chunk) - if digest.hexdigest() != asset["sha256"] or size != asset["size"]: - raise ValueError("附件内容不完整，ID：" + old_id) - info.update(suffix=suffix, storagePath=str(path), objName=path.name, - sizeKb=str(round(size / 1024, 2)), sizeInfo=str(size) + " B", delFlag="0") - prepared["sys_file"].append(info) + minio.check_bucket() + log("MinIO 桶连接成功：" + minio.bucket) + try: + for index, (old_id, info) in enumerate(file_info.items(), 1): + log("附件 [%s/%s] 开始上传及校验，源文件ID：%s" % (index, len(file_info), old_id)) + asset = manifest["files"][old_id] + digest, size = hashlib.sha256(), 0 + with tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024, mode="w+b") as body: + with archive.open(asset["member"]) as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + size += len(chunk) + body.write(chunk) + if digest.hexdigest() != asset["sha256"] or size != asset["size"]: + raise ValueError("附件内容不完整，ID：" + old_id) + body.seek(0) + content_type = mimetypes.guess_type(info["objName"])[0] or "application/octet-stream" + uploaded_location = "s3://" + minio.bucket + "/" + object_prefix + "/" + minio.upload_and_verify(info["storagePath"], body, size, asset["sha256"], content_type) + prepared["sys_file"].append(info) + log("附件 [%s/%s] 上传及 SHA-256 回读校验通过（%s 字节）" % (index, len(file_info), size)) + except (OSError, ValueError, RuntimeError, KeyError, zipfile.BadZipFile): + if uploaded_location: + print("MinIO 附件对象前缀（可能已有上传）：" + uploaded_location, file=sys.stderr) + raise + else: + log("导出包没有附件，跳过 MinIO 上传") sql = ["SET SESSION time_zone='+00:00';", "SET SESSION sql_mode='STRICT_ALL_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION';", "CREATE TEMPORARY TABLE inclusive_therapy_v3_import_guard (ok INT NOT NULL, CONSTRAINT inclusive_therapy_v3_import_requirement CHECK(ok=1)) ENGINE=InnoDB;", "START TRANSACTION;"] @@ -330,24 +481,33 @@ def main(): literal(version) + "," + literal("普惠疗休养V3报名规则版本") + ",0 WHERE NOT EXISTS " "(SELECT 1 FROM sys_config WHERE configKey='staffTherapyRecuperationVersion');") sql.append("COMMIT;") + log("[5/5] 开始数据库事务，写入 %s 条记录；数据量较大时请等待事务提交" % + sum(len(table_rows) for table_rows in prepared.values())) try: # mysql 批处理遇错即停止；连接关闭回滚未提交事务，不启用 --force。 for _ in client.lines("\n".join(sql)): pass except (OSError, RuntimeError): - if copied_directory: - print("附件暂存目录保留：" + str(copied_directory), file=sys.stderr) + if uploaded_location: + print("已上传附件保留在 MinIO 前缀：" + uploaded_location, file=sys.stderr) raise - print("导入完成：普惠疗休养V3，表前缀 " + PREFIX) - print("业务记录：%s；审核记录：%s；附件：%s" % - (sum(len(records[name]) for name in SUFFIXES), len(records["_audit"]), len(file_info))) - if copied_directory: - print("附件目录：" + str(copied_directory)) + log("数据库事务提交成功") + log("导入完成：普惠疗休养V3，表前缀 " + PREFIX) + log("业务记录：%s；审核记录：%s；附件：%s" % + (sum(len(records[name]) for name in SUFFIXES), len(records["_audit"]), len(file_info))) + if uploaded_location: + log("MinIO 附件对象前缀：" + uploaded_location) if __name__ == "__main__": + started = time.monotonic() try: main() + log("【成功结束】导入成功，总耗时 %.1f 秒，退出码 0" % (time.monotonic() - started)) + except KeyboardInterrupt: + log("【中断结束】用户中断导入，退出码 130；事务提交状态需确认，已上传的 MinIO 对象予以保留", error=True) + sys.exit(130) except (OSError, ValueError, RuntimeError, KeyError, zipfile.BadZipFile) as error: - print("导入失败：" + str(error), file=sys.stderr) + log("导入失败：" + str(error), error=True) + log("【失败结束】导入未成功，总耗时 %.1f 秒，退出码 1" % (time.monotonic() - started), error=True) sys.exit(1)