// Package syncer 编排监管中转主流程(与 xk-api HyTransitRecordService 配合)。 // // 每个 step(consult/referral/recipe/verification)在 runStep 中顺序执行: // // ① xkapi.BatchCreate → 云端 xk_hy_transit_batch 建批次 // ② xkapi.Pull → 云端组包并写入 xk_hy_transit_record,返回 record_id // ③ hy.PostForward(逐条) → 经 forward-go 报政务云;本地 hy_push_* 留痕 // ④ xkapi.BatchCallback → 整批回写 callback_status(非逐条 HTTP) // ⑤ xkapi.BatchFinish → 标记批次 pull 结束 // // 运营可见状态在云端表;本机表仅用于外网机审计与排错。 package syncer import ( "encoding/json" "fmt" "strings" "time" "xk-hy-transit-go/internal/applog" "xk-hy-transit-go/internal/config" "xk-hy-transit-go/internal/db" "xk-hy-transit-go/internal/hy" "xk-hy-transit-go/internal/hyfile" "xk-hy-transit-go/internal/xkapi" ) // Runner 单次或定时同步执行器。 type Runner struct { cfg config.Config xk *xkapi.Client store *db.Store organ hy.OrganConfig fileUploadURL string // 实际上传地址(经 forward 或直连政务云) uploadToken string // 由 xk-api config 接口生成的 uploadToken fileUploadViaForward bool } // StoreClose 关闭本机 MySQL 连接。 func (r *Runner) StoreClose() { if r.store != nil { _ = r.store.Close() } } // Store 返回本机审计库(供 Web 控制台只读查询)。 func (r *Runner) Store() *db.Store { return r.store } // NewRunner 加载机构配置并创建 Runner(含 forward 文件上传地址解析)。 func NewRunner(cfg config.Config, store *db.Store) (*Runner, error) { if p := strings.TrimSpace(cfg.ChromePath); p != "" { hyfile.SetChromePath(p) applog.Appf("chrome path from config: %s", p) } hyfile.SetUploadInsecureSkipVerify(cfg.FileUploadInsecureSkipVerify) xk := xkapi.New(cfg.XkAPIBaseURL, cfg.XkAPIToken, cfg.XkAPICallbackPath) conf, err := xk.Config() if err != nil { return nil, fmt.Errorf("load organ config: %w", err) } if strings.TrimSpace(conf.UploadToken) == "" { applog.Appf("warning: uploadToken empty, recipe file upload will fail (check xk-api HY_APP_KEY/HY_APP_SECRET/HY_FILE_BUCKET)") } uploadURL := resolveFileUploadURL(cfg, conf.FileUploadURL) if cfg.FileUploadViaForward { applog.Appf("file upload via forward: %s (cloud fileUploadUrl ignored)", uploadURL) } return &Runner{ cfg: cfg, xk: xk, store: store, organ: hy.OrganConfig{ UnitID: conf.UnitID, OrganID: conf.OrganID, OrganName: conf.OrganName, }, fileUploadURL: uploadURL, uploadToken: conf.UploadToken, fileUploadViaForward: cfg.FileUploadViaForward, }, nil } // resolveFileUploadURL 决定 PDF 实际上传地址:默认经 forward-go,避免外网机直连政务云 28211。 func resolveFileUploadURL(cfg config.Config, cloudFileUploadURL string) string { if cfg.FileUploadViaForward { return hy.ForwardFileUploadURL(cfg.ForwardBaseURL) } return strings.TrimSpace(cloudFileUploadURL) } // Run 执行一个或多个 step。 func (r *Runner) Run(step, anchorDate string) error { if step == "all" { for _, s := range []string{"consult", "referral", "recipe", "verification"} { if err := r.runStep(s, anchorDate); err != nil { return err } } return nil } return r.runStep(step, anchorDate) } // runStep 执行单个监管类型的完整同步(见包注释 ①~⑤)。 func (r *Runner) runStep(step, anchorDate string) error { method, ok := hy.StepMethods[step] if !ok { return fmt.Errorf("unknown step: %s", step) } applog.Appf("sync start step=%s date=%s", step, anchorDate) _, _ = r.store.UpsertJob(anchorDate, step, "running", 0, 0, 0, "") // ① 云端建批次 batchID, err := r.xk.BatchCreate(anchorDate, step) if err != nil { _ = r.store.FinishJob(anchorDate, step, "failed") applog.Pullf("batch create failed step=%s date=%s err=%v", step, anchorDate, err) return fmt.Errorf("batch create: %w", err) } applog.Pullf("batch created step=%s date=%s batch_id=%d", step, anchorDate, batchID) // ② 云端组包拉取(含 validation_errors、meta.needs_recipe_upload 等) pull, err := r.xk.Pull(step, anchorDate, batchID) if err != nil { _ = r.xk.BatchFinish(batchID, "failed") _ = r.store.FinishJob(anchorDate, step, "failed") applog.Pullf("pull failed step=%s batch_id=%d err=%v", step, batchID, err) return err } validationCount := 0 for _, item := range pull.Items { if len(item.ValidationErrors) > 0 { validationCount++ } } applog.Pullf("pull done step=%s batch_id=%d items=%d validation_failed=%d", step, batchID, len(pull.Items), validationCount) var callbacks []xkapi.CallbackItem success, failed, skipped := 0, 0, 0 // ③ 逐条:recipe 可能先 PDF 上传,再加密经 forward 报政务云 for _, item := range pull.Items { cb := r.processItem(step, anchorDate, method, batchID, item) callbacks = append(callbacks, cb) switch cb.PushStatus { case "success": success++ case "skipped": skipped++ default: failed++ } } // ④ 整批回调云端(写 push_status / callback_status) if err := r.xk.BatchCallback(batchID, callbacks); err != nil { applog.Appf("batch callback warning batch_id=%d err=%v", batchID, err) } // ⑤ 标记批次结束 _ = r.xk.BatchFinish(batchID, "done") total := success + failed + skipped _, _ = r.store.UpsertJob(anchorDate, step, "done", total, success, failed, "") _ = r.store.FinishJob(anchorDate, step, "done") applog.Pullf("step done step=%s batch_id=%d total=%d success=%d failed=%d skipped=%d", step, batchID, total, success, failed, skipped) return nil } // processItem 处理单条:校验失败则 skipped;否则 BuildUpload + PostForward,组装 CallbackItem。 func (r *Runner) processItem(step, anchorDate, method string, batchID int, item xkapi.PullItem) xkapi.CallbackItem { payload := item.Payload bizKey := item.BizKey if bizKey == "" && payload != nil { bizKey = hy.BizKey(step, anchorDate, payload) } recordID := item.RecordID cb := xkapi.CallbackItem{ RecordID: recordID, BizKey: bizKey, CallbackStatus: "waiting", } if len(item.ValidationErrors) > 0 || payload == nil { bussID := "" if payload != nil { bussID = fmt.Sprintf("%v", payload["bussID"]) } payloadJSON, _ := json.Marshal(payload) _ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, bussID, bizKey, "", string(payloadJSON), item.ValidationErrors) cb.PushStatus = "skipped" cb.CallbackStatus = "failed" cb.ErrorMessage = "validation failed" if len(item.ValidationErrors) > 0 { cb.ErrorMessage = item.ValidationErrors[0] } applog.Pullf("skip validation biz_key=%s errors=%v", bizKey, item.ValidationErrors) return cb } // recipe 且需上传处方 PDF:先经 forward 拿到 recipeFileId,再加密上报监管 if step == "recipe" && metaBool(item.Meta, "needs_recipe_upload") { if err := r.ensureRecipeFileID(item, payload); err != nil { _ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", mustJSON(payload), nil) cb.PushStatus = "failed" cb.CallbackStatus = "failed" cb.ErrorMessage = "recipe file upload: " + err.Error() applog.Pushf("recipe upload fail biz_key=%s err=%v", bizKey, err) return cb } } payloadJSON, _ := json.Marshal(payload) _ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", string(payloadJSON), nil) req, err := hy.BuildUpload(method, payload, r.organ, r.cfg.HyAppKey, r.cfg.HyAppSecret, r.cfg.HyAesKey) if err != nil { _ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error()) cb.PushStatus = "failed" cb.CallbackStatus = "failed" cb.ErrorMessage = err.Error() applog.Pushf("build upload fail biz_key=%s err=%v", bizKey, err) return cb } start := time.Now() result, err := hy.PostForward(r.cfg.ForwardBaseURL, req, r.cfg.ForwardSharedSecret) duration := int(time.Since(start).Milliseconds()) if err != nil { _ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error()) cb.PushStatus = "failed" cb.CallbackStatus = "failed" cb.ErrorMessage = err.Error() applog.Pushf("post forward fail biz_key=%s method=%s err=%v", bizKey, method, err) return cb } recordLocalID, _ := r.store.GetRecordID(bizKey) headersJSON, _ := json.Marshal(req.Headers) _ = r.store.SaveLog(recordLocalID, result.HTTPCode, result.MsgCode, result.Msg, result.TraceID, len(req.Body), req.PlainJSON, string(headersJSON), result.Body, duration) cb.HTTPCode = result.HTTPCode cb.MsgCode = result.MsgCode cb.Msg = result.Msg cb.TraceID = result.TraceID cb.ResponseBody = result.Body if hy.IsSuperviseOK(result) { _ = r.store.UpdateRecordStatus(bizKey, "success", "") cb.PushStatus = "success" cb.CallbackStatus = "success" applog.Pushf("ok biz_key=%s method=%s code=%d msgCode=%d trace=%s %dms", bizKey, method, result.GatewayCode, result.MsgCode, result.TraceID, duration) } else { errText := hy.SuperviseErrorSummary(result) _ = r.store.UpdateRecordStatus(bizKey, "failed", errText) cb.PushStatus = "failed" cb.CallbackStatus = "failed" cb.ErrorMessage = errText applog.Pushf("fail biz_key=%s method=%s %s %dms", bizKey, method, errText, duration) } return cb } // ResolveAnchorDate 解析锚定日:命令行优先,否则按配置向前推 N 天。 func ResolveAnchorDate(cfg config.Config, dateArg string) string { if dateArg != "" { return dateArg } return time.Now().AddDate(0, 0, -cfg.AnchorOffsetDays).Format("2006-01-02") } // ensureRecipeFileID 处方 PDF 子流程:HTML→PDF→本地落盘→经 forward 上传→写回 recipeFileId。 func (r *Runner) ensureRecipeFileID(item xkapi.PullItem, payload map[string]any) error { // 步骤 1:解析处方 ID,必要时从云端拉取打印 HTML prescriptionID := metaInt(item.Meta, "prescription_id") if prescriptionID <= 0 { prescriptionID = metaIntFromPayload(payload, "recipeID") } html, _ := item.Meta["recipe_file_html"].(string) if strings.TrimSpace(html) == "" { if prescriptionID <= 0 { return fmt.Errorf("recipe_file_html missing and no prescription_id") } applog.Pushf("fetch recipe html prescription_id=%d", prescriptionID) detail, err := r.xk.GetPrescriptionPrintDetail(prescriptionID) if err != nil { return fmt.Errorf("fetch recipe html: %w", err) } html = detail.RecipeFileHTML } // 步骤 2:chromedp 将监管打印 HTML 转为 PDF 字节 applog.Pushf("recipe pdf: prescription_id=%d html_len=%d save_root=%s", prescriptionID, len(html), hyfile.ProgramDir()) pdf, err := hyfile.HtmlToPDF(html) if err != nil { return err } // 步骤 3:落盘到本机 pdf/ 目录,便于运营核对 pdfName := buildPDFNameInput(item.Meta, payload) var localPath string if saved, err := hyfile.SavePDFLocal(pdf, pdfName); err != nil { applog.Pushf("save local pdf warning prescription_id=%d err=%v", prescriptionID, err) } else { localPath = saved applog.Pushf("saved local pdf: %s", saved) } // 步骤 4:经 forward-go 上传至政务云文件服务,得到 recipeFileId filename := hyfile.BuildPDFFilename(pdfName) fileID, err := hyfile.UploadPDF(pdf, filename, r.fileUploadURL, r.uploadToken, r.cfg.ForwardSharedSecret) if err != nil { applog.Pushf("upload pdf fail prescription_no=%s local=%s err=%v", pdfName.PrescriptionNo, localPath, err) return err } applog.Pushf("upload pdf ok prescription_no=%s fileId=%s local=%s url=%s", pdfName.PrescriptionNo, fileID, localPath, r.fileUploadURL) // 步骤 5:写入组包 payload,并回写云端处方表 payload["recipeFileId"] = fileID if prescriptionID > 0 { if err := r.xk.SaveRecipeFile(prescriptionID, fileID, item.RecordID); err != nil { applog.Pushf("save recipe file cloud warning prescription_id=%d err=%v", prescriptionID, err) } } return nil } func metaBool(meta map[string]any, key string) bool { if meta == nil { return false } switch v := meta[key].(type) { case bool: return v case float64: return v != 0 case int: return v != 0 case string: s := strings.TrimSpace(strings.ToLower(v)) return s == "true" || s == "1" || s == "yes" default: return false } } func metaString(meta map[string]any, key string) string { if meta == nil { return "" } if s, ok := meta[key].(string); ok { return s } return fmt.Sprintf("%v", meta[key]) } func metaInt(meta map[string]any, key string) int { if meta == nil { return 0 } switch v := meta[key].(type) { case float64: return int(v) case int: return v case string: var n int fmt.Sscanf(v, "%d", &n) return n default: return 0 } } func buildPDFNameInput(meta map[string]any, payload map[string]any) hyfile.PDFNameInput { in := hyfile.PDFNameInput{At: parseMetaTime(meta, payload)} in.PatientName = metaString(meta, "patient_name") if in.PatientName == "" { if s, ok := payload["patientName"].(string); ok { in.PatientName = s } } in.StoreName = metaString(meta, "store_name") in.PrescriptionNo = metaString(meta, "prescription_no") if in.PrescriptionNo == "" { in.PrescriptionNo = fmt.Sprintf("%v", payload["recipeUniqueID"]) } return in } func parseMetaTime(meta map[string]any, payload map[string]any) time.Time { if meta != nil { if t := hyfile.ParseTimeString(metaString(meta, "created_at")); !t.IsZero() { return t } } if s, ok := payload["datein"].(string); ok { if t := hyfile.ParseTimeString(s); !t.IsZero() { return t } } return time.Now() } func metaIntFromPayload(payload map[string]any, key string) int { switch v := payload[key].(type) { case float64: return int(v) case int: return v case string: var n int fmt.Sscanf(v, "%d", &n) return n default: return 0 } } func mustJSON(payload map[string]any) string { b, _ := json.Marshal(payload) return string(b) } // FileUploadURL 返回实际使用的文件上传地址(upload-test 与排错用)。 func (r *Runner) FileUploadURL() string { return r.fileUploadURL } // FileUploadViaForward 是否经内网 forward 上传 PDF。 func (r *Runner) FileUploadViaForward() bool { return r.fileUploadViaForward } // UploadToken 返回监管文件上传凭证。 func (r *Runner) UploadToken() string { return r.uploadToken } // ForwardSharedSecret 与 forward-go FORWARD_SHARED_SECRET 一致(可选)。 func (r *Runner) ForwardSharedSecret() string { return r.cfg.ForwardSharedSecret }