// 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" "sync" "time" "xk-hy-transit-go/internal/applog" "xk-hy-transit-go/internal/config" "xk-hy-transit-go/internal/db" "xk-hy-transit-go/internal/fileauth" "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 // 本地按 FileAuth.java 生成 cloudUploadToken string // xk-api 预生成值,仅诊断对比 fileBucket string fileUploadViaForward bool hyAppKey string hyAppSecret string hyAesKey string lastCloud xkapi.ConfigResponse lastCloudOK bool lastConfigErr error configRefreshedAt time.Time configMu sync.RWMutex } // 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) { fileauth.SetB64Mode(cfg.FileAuthB64) if p := strings.TrimSpace(cfg.ChromePath); p != "" { hyfile.SetChromePath(p) applog.Appf("chrome path from config: %s", p) } hyfile.SetUploadInsecureSkipVerify(cfg.FileUploadInsecureSkipVerify) hyfile.SetUploadMinInterval(cfg.FileUploadMinInterval) if cfg.FileUploadMinInterval > 0 { applog.Appf("file upload min interval: %s", cfg.FileUploadMinInterval) } xk := xkapi.New(cfg.XkAPIBaseURL, cfg.XkAPIToken, cfg.XkAPICallbackPath) r := &Runner{ cfg: cfg, xk: xk, store: store, fileUploadViaForward: cfg.FileUploadViaForward, hyAppKey: cfg.HyAppKey, hyAppSecret: cfg.HyAppSecret, hyAesKey: cfg.HyAesKey, } if err := r.ReloadCloudConfig(); err != nil { return nil, fmt.Errorf("load organ config: %w", err) } if strings.TrimSpace(r.uploadToken) == "" { applog.Appf("warning: uploadToken empty, recipe file upload will fail (check xk-api HY_APP_KEY/HY_APP_SECRET/HY_FILE_BUCKET)") } if cfg.FileUploadViaForward { applog.Appf("file upload via forward: %s (cloud fileUploadUrl ignored)", r.fileUploadURL) } return r, 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 { _, err := r.executeStep(step, anchorDate, StepRunOptions{ItemLimit: 0, CollectTrace: false}) return err } // executeStep 单 step 完整链路;自动与联调共用,开头刷新云端配置与 uploadToken。 func (r *Runner) executeStep(step, anchorDate string, opts StepRunOptions) (*StepTestResult, error) { method, ok := hy.StepMethods[step] if !ok { return nil, fmt.Errorf("unknown step: %s", step) } if err := r.ReloadCloudConfig(); err != nil { return nil, fmt.Errorf("reload config: %w", err) } var result *StepTestResult if opts.CollectTrace { result = &StepTestResult{ Step: step, StepLabel: hy.LabelStep(step), AnchorDate: anchorDate, } } applog.Appf("sync start step=%s date=%s", step, anchorDate) if r.store != nil { _, _ = r.store.UpsertJob(anchorDate, step, "running", 0, 0, 0, "") } batchID, err := r.xk.BatchCreate(anchorDate, step) if err != nil { if r.store != nil { _ = r.store.FinishJob(anchorDate, step, "failed") } applog.Pullf("batch create failed step=%s date=%s err=%v", step, anchorDate, err) return result, fmt.Errorf("batch create: %w", err) } if result != nil { result.BatchID = batchID } applog.Pullf("batch created step=%s date=%s batch_id=%d", step, anchorDate, batchID) pull, err := r.xk.Pull(step, anchorDate, batchID) if err != nil { _ = r.xk.BatchFinish(batchID, "failed") if r.store != nil { _ = r.store.FinishJob(anchorDate, step, "failed") } applog.Pullf("pull failed step=%s batch_id=%d err=%v", step, batchID, err) return result, err } validationCount := 0 for _, item := range pull.Items { if len(item.ValidationErrors) > 0 { validationCount++ } } followUpCount := 0 if step == "verification" { followUpCount = len(pull.FollowUpRecipeItems) } applog.Pullf("pull done step=%s batch_id=%d items=%d validation_failed=%d follow_up_recipe=%d", step, batchID, len(pull.Items), validationCount, followUpCount) if result != nil { result.PullSummary = PullSummary{ Total: len(pull.Items), ValidationFailed: validationCount, } } items := pull.Items if opts.ItemLimit > 0 && len(items) > opts.ItemLimit { items = items[:opts.ItemLimit] } if result != nil { result.PullSummary.Processed = len(items) } var callbacks []xkapi.CallbackItem success, failed, skipped := 0, 0, 0 appendProcessed := func(processStep, processMethod string, batchItems []xkapi.PullItem) { for _, item := range batchItems { var tr *itemTraceCtx if opts.CollectTrace { tr = newItemTraceCtx(item) if item.BizKey == "" && item.Payload != nil { tr.out.BizKey = hy.BizKey(processStep, anchorDate, item.Payload) } } cb := r.processItem(processStep, anchorDate, processMethod, batchID, item, tr) callbacks = append(callbacks, cb) if tr != nil { tr.finish(cb) result.Items = append(result.Items, *tr.out) } switch cb.PushStatus { case "success": success++ case "skipped": skipped++ default: failed++ } } } appendProcessed(step, method, items) if step == "verification" && len(pull.FollowUpRecipeItems) > 0 { followUpMethod, ok := hy.StepMethods["recipe"] if !ok { return result, fmt.Errorf("unknown follow-up step: recipe") } applog.Pullf("verification follow-up recipe items=%d", len(pull.FollowUpRecipeItems)) appendProcessed("recipe", followUpMethod, pull.FollowUpRecipeItems) } if err := r.xk.BatchCallback(batchID, callbacks); err != nil { applog.Appf("batch callback failed step=%s batch_id=%d err=%v", step, batchID, err) if r.store != nil { _, _ = r.store.UpsertJob(anchorDate, step, "failed", success+failed+skipped, success, failed, err.Error()) _ = r.store.FinishJob(anchorDate, step, "failed") } if result != nil { result.CallbackError = err.Error() result.Summary = StepOutcomeSummary{Success: success, Failed: failed, Skipped: skipped} } _ = r.xk.BatchFinish(batchID, "failed") return result, fmt.Errorf("batch callback: %w", err) } _ = r.xk.BatchFinish(batchID, "done") total := success + failed + skipped if r.store != nil { _, _ = 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) if result != nil { result.Summary = StepOutcomeSummary{Success: success, Failed: failed, Skipped: skipped} } return result, nil } // processItem 处理单条:校验失败则 skipped;否则 BuildUpload + PostForward,组装 CallbackItem。 func (r *Runner) processItem(step, anchorDate, method string, batchID int, item xkapi.PullItem, tr *itemTraceCtx) 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) if r.store != nil { _ = 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 } if tr != nil { tr.setPayloadSnapshot(payload) } if step == "recipe" { if metaBool(item.Meta, "needs_recipe_upload") { if err := r.ensureRecipeFileID(item, payload, tr); err != nil { if r.store != 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 } } else if tr != nil { tr.setRecipeSkipped("needs_recipe_upload=false") } } payloadJSON, _ := json.Marshal(payload) if r.store != nil { _ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", string(payloadJSON), nil) } r.configMu.RLock() appKey, appSecret, aesKey := r.hyAppKey, r.hyAppSecret, r.hyAesKey r.configMu.RUnlock() req, err := hy.BuildUpload(method, payload, r.organ, appKey, appSecret, aesKey) if err != nil { if r.store != 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 } if tr != nil { tr.setPushReq(req) } start := time.Now() result, err := hy.PostForward(r.cfg.ForwardBaseURL, req, r.cfg.ForwardSharedSecret) duration := int(time.Since(start).Milliseconds()) if err != nil { if r.store != 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 } if r.store != nil { 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) { if r.store != nil { _ = 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) if r.store != nil { _ = 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, tr *itemTraceCtx) 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(§2.3.4 multipart 字段 file) filename := hyfile.BuildPDFFilename(pdfName) token, err := r.uploadTokenForFile(filename) if err != nil { return err } r.configMu.RLock() url := r.fileUploadURL r.configMu.RUnlock() upRes, err := hyfile.UploadPDFDetail(pdf, filename, url, token, r.cfg.ForwardSharedSecret) if tr != nil { tr.setRecipeUpload(upRes, localPath, err) } if err != nil { applog.Pushf("upload pdf fail prescription_no=%s local=%s err=%v", pdfName.PrescriptionNo, localPath, err) return err } fileID := upRes.FileID 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 } // Cfg 返回本地环境配置(只读副本)。 func (r *Runner) Cfg() config.Config { return r.cfg } // FileBucket 返回生效的文件 bucket(uploadToken scope)。 func (r *Runner) FileBucket() string { r.configMu.RLock() defer r.configMu.RUnlock() return r.fileBucket } // CloudUploadToken 返回 xk-api 预生成的 uploadToken(仅诊断对比)。 func (r *Runner) CloudUploadToken() string { r.configMu.RLock() defer r.configMu.RUnlock() return r.cloudUploadToken } // UploadTokenSource 说明 uploadToken 生成方式。 func (r *Runner) UploadTokenSource() string { return "local-fileauth-java" } // RegenerateUploadToken 供诊断:用当前凭证重新生成 token(与 refreshUploadToken 相同)。 func (r *Runner) RegenerateUploadToken() error { return r.refreshUploadToken() } // RetryRecord 对本机 hy_push_record 整条重试:刷新配置 → 处理 → 回调云端。 func (r *Runner) RetryRecord(localRecordID int64) error { if r.store == nil { return fmt.Errorf("no local store") } if err := r.ReloadCloudConfig(); err != nil { return fmt.Errorf("reload config: %w", err) } detail, err := r.store.GetRecordDetail(localRecordID) if err != nil { return err } rec := detail.Record step := db.StepForMethod(rec.Method) if step == "" { return fmt.Errorf("unknown method: %s", rec.Method) } method := rec.Method batchID := int(rec.JobID) if batchID <= 0 { return fmt.Errorf("invalid batch_id (job_id) on record %d", localRecordID) } var payload map[string]any if rec.PayloadJSON != "" { if err := json.Unmarshal([]byte(rec.PayloadJSON), &payload); err != nil { return fmt.Errorf("parse payload_json: %w", err) } } var validationErrors []string if rec.ValidationErrors.Valid && rec.ValidationErrors.String != "" { _ = json.Unmarshal([]byte(rec.ValidationErrors.String), &validationErrors) } meta := buildRetryMeta(step, payload) item := xkapi.PullItem{ RecordID: 0, BizKey: rec.BizKey, Payload: payload, Meta: meta, ValidationErrors: validationErrors, } cb := r.processItem(step, rec.AnchorDate, method, batchID, item, nil) if err := r.xk.BatchCallback(batchID, []xkapi.CallbackItem{cb}); err != nil { applog.Appf("retry batch callback failed record_id=%d err=%v", localRecordID, err) return fmt.Errorf("batch callback: %w", err) } _ = r.store.IncRetry(rec.BizKey) if cb.PushStatus != "success" { if cb.ErrorMessage != "" { return fmt.Errorf("%s", cb.ErrorMessage) } return fmt.Errorf("retry failed push_status=%s", cb.PushStatus) } return nil } func buildRetryMeta(step string, payload map[string]any) map[string]any { meta := map[string]any{} if step != "recipe" || payload == nil { return meta } rid := metaIntFromPayload(payload, "recipeID") if rid > 0 { meta["prescription_id"] = rid } if v, ok := payload["recipeFileId"]; !ok || v == nil || fmt.Sprintf("%v", v) == "" { meta["needs_recipe_upload"] = true } return meta }