From 16aebb45b6bec285c310c544b1a3ee2044c48d11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E7=90=A6?= Date: Fri, 5 Jun 2026 16:36:39 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=A4=84=E6=96=B9=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/hyfile/upload.go | 87 +++++++++---- internal/hyfile/upload_detail_test.go | 53 ++++++++ internal/logweb/server.go | 2 + internal/logweb/supervise_handlers.go | 68 ++++++++++ internal/logweb/web/app.js | 1 + internal/logweb/web/index.html | 4 + internal/sync/step_result.go | 159 +++++++++++++++++++++++ internal/sync/step_test_run_test.go | 14 ++ internal/sync/supervise_step.go | 23 ++++ internal/sync/sync.go | 176 ++++++++++++++++++++------ 10 files changed, 523 insertions(+), 64 deletions(-) create mode 100644 internal/hyfile/upload_detail_test.go create mode 100644 internal/logweb/supervise_handlers.go create mode 100644 internal/sync/step_result.go create mode 100644 internal/sync/step_test_run_test.go create mode 100644 internal/sync/supervise_step.go diff --git a/internal/hyfile/upload.go b/internal/hyfile/upload.go index 87f6b24..39f35a2 100644 --- a/internal/hyfile/upload.go +++ b/internal/hyfile/upload.go @@ -21,6 +21,7 @@ import ( const uploadFormFieldFile = "file" const defaultUploadFilename = "prescription.pdf" +const uploadRawBodyMaxLen = 4096 var ( uploadInsecureSkipVerify bool @@ -52,6 +53,27 @@ func uploadHTTPClient() *http.Client { return &http.Client{Timeout: 120 * time.Second, Transport: transport} } +// UploadPDFResult 文件上传 HTTP 结果（供联调展示 raw 响应）。 +type UploadPDFResult struct { + FileID string `json:"fileId,omitempty"` + HTTPStatus int `json:"httpStatus"` + RawBody string `json:"rawBody"` + Success bool `json:"success"` + Message string `json:"message,omitempty"` +} + +// UploadPDFDetail 上传 PDF 并返回 fileId 与原始响应（与 UploadPDF 同一 HTTP 实现）。 +func UploadPDFDetail(pdf []byte, filename, uploadURL, uploadToken, forwardToken string) (*UploadPDFResult, error) { + res, err := doUploadPDF(pdf, filename, uploadURL, uploadToken, forwardToken) + if err != nil { + if res != nil { + return res, err + } + return nil, err + } + return res, nil +} + // UploadPDF 上传 PDF 至监管文件服务（通常经 forward-go），成功时返回 record.fileId。 // // uploadURL：FILE_UPLOAD_VIA_FORWARD=true 时为 forward 地址；否则为云端 config 返回的政务云 URL。 @@ -59,24 +81,32 @@ func uploadHTTPClient() *http.Client { // filename：multipart 表单 file 字段文件名；空时默认 prescription.pdf。 // forwardToken：非空时设置 X-Forward-Token（与 forward-go FORWARD_SHARED_SECRET 对应）。 func UploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string) (string, error) { + res, err := doUploadPDF(pdf, filename, uploadURL, uploadToken, forwardToken) + if err != nil { + return "", err + } + return res.FileID, nil +} + +func doUploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string) (*UploadPDFResult, error) { if len(pdf) == 0 { - return "", fmt.Errorf("empty pdf") + return nil, fmt.Errorf("empty pdf") } if uploadURL == "" { - return "", fmt.Errorf("file upload url empty") + return nil, fmt.Errorf("file upload url empty") } if uploadToken == "" { - return "", fmt.Errorf("upload token empty") + return nil, fmt.Errorf("upload token empty") } contentType, body, err := buildMultipartPDFBody(pdf, filename) if err != nil { - return "", err + return nil, err } req, err := http.NewRequest(http.MethodPost, uploadURL, body) if err != nil { - return "", err + return nil, err } req.Header.Set("Content-Type", contentType) req.Header.Set("X-Authorization", uploadToken) @@ -87,15 +117,39 @@ func UploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string acquireUploadSlot() resp, err := uploadHTTPClient().Do(req) if err != nil { - return "", err + return nil, err } defer resp.Body.Close() raw, _ := io.ReadAll(resp.Body) - - if resp.StatusCode != http.StatusOK { - return "", formatUploadHTTPError(resp.StatusCode, raw, uploadURL) + out := &UploadPDFResult{ + HTTPStatus: resp.StatusCode, + RawBody: string(truncate(raw, uploadRawBodyMaxLen)), } + if resp.StatusCode != http.StatusOK { + return out, formatUploadHTTPError(resp.StatusCode, raw, uploadURL) + } + + fileID, success, msg, parseErr := parseUploadResponseBody(raw) + out.Success = success + out.Message = msg + out.FileID = fileID + if parseErr != nil { + return out, parseErr + } + if !success { + if msg == "" { + msg = out.RawBody + } + return out, fmt.Errorf("upload failed: %s", msg) + } + if fileID == "" { + return out, fmt.Errorf("upload response missing record.fileId: %s", truncate(raw, 512)) + } + return out, nil +} + +func parseUploadResponseBody(raw []byte) (fileID string, success bool, message string, err error) { var parsed struct { Success bool `json:"success"` Record struct { @@ -104,20 +158,9 @@ func UploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string Message string `json:"message"` } if err := json.Unmarshal(raw, &parsed); err != nil { - return "", fmt.Errorf("parse upload response: %w", err) + return "", false, "", fmt.Errorf("parse upload response: %w", err) } - if !parsed.Success { - msg := parsed.Message - if msg == "" { - msg = string(truncate(raw, 512)) - } - return "", fmt.Errorf("upload failed: %s", msg) - } - fileID := parsed.Record.FileID - if fileID == "" { - return "", fmt.Errorf("upload response missing record.fileId: %s", truncate(raw, 512)) - } - return fileID, nil + return parsed.Record.FileID, parsed.Success, parsed.Message, nil } func buildMultipartPDFBody(pdf []byte, filename string) (contentType string, body io.Reader, err error) { diff --git a/internal/hyfile/upload_detail_test.go b/internal/hyfile/upload_detail_test.go new file mode 100644 index 0000000..cb1bfb4 --- /dev/null +++ b/internal/hyfile/upload_detail_test.go @@ -0,0 +1,53 @@ +package hyfile + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" +) + +func TestUploadPDFDetail_returnsRawBody(t *testing.T) { + disableUploadThrottle(t) + rawResp := `{"success":true,"record":{"fileId":"fid-99"},"message":"ok"}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + _, _ = w.Write([]byte(rawResp)) + })) + defer srv.Close() + + res, err := UploadPDFDetail([]byte("%PDF-test"), "a.pdf", srv.URL, "tok", "") + if err != nil { + t.Fatal(err) + } + if res.FileID != "fid-99" || !res.Success { + t.Fatalf("res=%+v", res) + } + if res.RawBody != rawResp { + t.Fatalf("raw=%q", res.RawBody) + } + if res.HTTPStatus != http.StatusOK { + t.Fatalf("status=%d", res.HTTPStatus) + } +} + +func TestUploadPDFDetail_httpErrorPreservesBody(t *testing.T) { + disableUploadThrottle(t) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusForbidden) + _ = json.NewEncoder(w).Encode(map[string]string{"message": "denied"}) + })) + defer srv.Close() + + res, err := UploadPDFDetail([]byte("x"), "x.pdf", srv.URL, "tok", "") + if err == nil { + t.Fatal("expected error") + } + if res == nil || res.HTTPStatus != 403 { + t.Fatalf("res=%+v err=%v", res, err) + } + if res.RawBody == "" { + t.Fatal("expected raw body") + } +} diff --git a/internal/logweb/server.go b/internal/logweb/server.go index 654b873..55f9356 100644 --- a/internal/logweb/server.go +++ b/internal/logweb/server.go @@ -67,6 +67,7 @@ func (s *Server) routes() { s.mux.HandleFunc("/runs", s.handlePage("runs.html")) s.mux.HandleFunc("/runs/record", s.handlePage("runs_record.html")) s.mux.HandleFunc("/test", s.handlePage("test.html")) + s.mux.HandleFunc("/supervise-step", s.handleSuperviseStepPage) s.mux.HandleFunc("/config", s.handlePage("config.html")) s.mux.HandleFunc("/upload", s.handlePage("upload.html")) s.mux.HandleFunc("/fileauth", s.handlePage("fileauth.html")) @@ -84,6 +85,7 @@ func (s *Server) routes() { s.mux.HandleFunc("/api/runs/retry", s.handleAPIRunsRetry) s.mux.HandleFunc("/api/test/status", s.handleAPITestStatus) s.mux.HandleFunc("/api/test/sync", s.handleAPITestSync) + s.mux.HandleFunc("/api/supervise/step-test", s.handleAPISuperviseStepTest) s.mux.Handle("/static/", http.StripPrefix("/static/", fileServer)) } diff --git a/internal/logweb/supervise_handlers.go b/internal/logweb/supervise_handlers.go new file mode 100644 index 0000000..736ffd8 --- /dev/null +++ b/internal/logweb/supervise_handlers.go @@ -0,0 +1,68 @@ +package logweb + +import ( + "encoding/json" + "net/http" + "strings" + + syncer "xk-hy-transit-go/internal/sync" + "xk-hy-transit-go/internal/syncgate" +) + +func (s *Server) handleSuperviseStepPage(w http.ResponseWriter, r *http.Request) { + s.handlePage("supervise_step.html")(w, r) +} + +func (s *Server) handleAPISuperviseStepTest(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeErr(w, 405, "POST only") + return + } + sr := asSyncRunner(s.deps.Runner) + if sr == nil { + writeErr(w, 503, "未配置 Runner（请使用 transit serve）") + return + } + var body struct { + Step string `json:"step"` + Date string `json:"date"` + Limit *int `json:"limit"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeErr(w, 400, "invalid json") + return + } + step := strings.TrimSpace(body.Step) + if step == "" { + writeErr(w, 400, "step required (recipe|verification)") + return + } + limit := 1 + if body.Limit != nil { + limit = *body.Limit + } + anchor := syncer.ResolveAnchorDate(s.deps.Cfg, strings.TrimSpace(body.Date)) + + var result *syncer.StepTestResult + var runErr error + if !syncgate.TryRun(step, anchor, func() error { + result, runErr = sr.TestSuperviseStep(step, anchor, limit) + return runErr + }) { + writeErr(w, 409, "已有同步任务在执行中") + return + } + if runErr != nil { + payload := map[string]any{"error": runErr.Error()} + if result != nil { + payload["result"] = result + } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(502) + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = enc.Encode(payload) + return + } + writeJSON(w, result) +} diff --git a/internal/logweb/web/app.js b/internal/logweb/web/app.js index 11caab2..c2b8b65 100644 --- a/internal/logweb/web/app.js +++ b/internal/logweb/web/app.js @@ -102,6 +102,7 @@ const MAIN_NAV = [ { href: '/config', id: 'config', label: '配置' }, { href: '/fileauth', id: 'fileauth', label: '上传凭证' }, { href: '/upload', id: 'upload', label: '上传' }, + { href: '/supervise-step', id: 'supervise-step', label: '处方核销' }, { href: '/runs', id: 'runs', label: '流水' }, { href: '/logs', id: 'logs', label: '日志' }, { href: '/test', id: 'test', label: '测试' }, diff --git a/internal/logweb/web/index.html b/internal/logweb/web/index.html index 1539072..d5816a5 100644 --- a/internal/logweb/web/index.html +++ b/internal/logweb/web/index.html @@ -31,6 +31,10 @@

文件日志

查看 app / pull / push 按日滚动日志，支持搜索与高亮。

+ +

处方/核销联调

+

在线处方与处方核销单步测试：组包、文件上传 raw、上报、云端回调。

+

测试执行

触发一次 sync（拉取 → 上传 → 回调），用于联调测试。

diff --git a/internal/sync/step_result.go b/internal/sync/step_result.go new file mode 100644 index 0000000..3b091ea --- /dev/null +++ b/internal/sync/step_result.go @@ -0,0 +1,159 @@ +package syncer + +import ( + "xk-hy-transit-go/internal/hy" + "xk-hy-transit-go/internal/hyfile" + "xk-hy-transit-go/internal/xkapi" +) + +// StepRunOptions 单 step 执行选项（自动 sync 与联调共用 executeStep）。 +type StepRunOptions struct { + ItemLimit int // 0=处理 Pull 全部条数；>0 只处理前 N 条 + CollectTrace bool // true 时填充 StepTestResult +} + +// StepTestResult 联调页返回（recipe / verification）。 +type StepTestResult struct { + Step string `json:"step"` + StepLabel string `json:"step_label"` + AnchorDate string `json:"anchor_date"` + BatchID int `json:"batch_id"` + PullSummary PullSummary `json:"pull_summary"` + Items []ItemStepTrace `json:"items"` + CallbackError string `json:"callback_error,omitempty"` + Summary StepOutcomeSummary `json:"summary"` +} + +// PullSummary 拉取摘要。 +type PullSummary struct { + Total int `json:"total"` + ValidationFailed int `json:"validation_failed"` + Processed int `json:"processed"` +} + +// StepOutcomeSummary 本 step 处理结果统计。 +type StepOutcomeSummary struct { + Success int `json:"success"` + Failed int `json:"failed"` + Skipped int `json:"skipped"` +} + +// ItemStepTrace 单条联调追踪（与 processItem 一一对应）。 +type ItemStepTrace struct { + RecordID int `json:"record_id"` + BizKey string `json:"biz_key"` + ValidationErrors []string `json:"validation_errors,omitempty"` + PayloadSnapshot string `json:"payload_snapshot,omitempty"` + PushPlainJSON string `json:"push_plain_json,omitempty"` + PushHeaders map[string]string `json:"push_headers,omitempty"` + RecipeFileUpload *RecipeFileUploadTrace `json:"recipe_file_upload,omitempty"` + Push PushTrace `json:"push"` + Callback xkapi.CallbackItem `json:"callback"` +} + +// RecipeFileUploadTrace 处方 PDF 28211 上传追踪。 +type RecipeFileUploadTrace struct { + Executed bool `json:"executed"` + Skipped bool `json:"skipped"` + SkipReason string `json:"skip_reason,omitempty"` + FileID string `json:"fileId,omitempty"` + HTTPStatus int `json:"httpStatus,omitempty"` + RawBody string `json:"rawBody,omitempty"` + Success bool `json:"success"` + Message string `json:"message,omitempty"` + LocalPDFPath string `json:"localPdfPath,omitempty"` + Error string `json:"error,omitempty"` +} + +// PushTrace 监管 JSON 上报结果摘要。 +type PushTrace struct { + HTTPCode int `json:"http_code,omitempty"` + MsgCode int `json:"msg_code,omitempty"` + Msg string `json:"msg,omitempty"` + TraceID string `json:"trace_id,omitempty"` + ResponseBody string `json:"response_body,omitempty"` +} + +// itemTraceCtx processItem 内部填充，最终写入 ItemStepTrace。 +type itemTraceCtx struct { + out *ItemStepTrace +} + +func newItemTraceCtx(item xkapi.PullItem) *itemTraceCtx { + return &itemTraceCtx{out: &ItemStepTrace{ + RecordID: item.RecordID, + BizKey: item.BizKey, + ValidationErrors: append([]string(nil), item.ValidationErrors...), + }} +} + +func (c *itemTraceCtx) setPayloadSnapshot(payload map[string]any) { + if c == nil || c.out == nil || payload == nil { + return + } + c.out.PayloadSnapshot = mustJSON(payload) +} + +func (c *itemTraceCtx) setPushReq(req *hy.UploadRequest) { + if c == nil || c.out == nil || req == nil { + return + } + c.out.PushPlainJSON = req.PlainJSON + c.out.PushHeaders = copyHeadersForTrace(req.Headers) +} + +func copyHeadersForTrace(h map[string]string) map[string]string { + if h == nil { + return nil + } + out := make(map[string]string, len(h)) + for k, v := range h { + if k == "secret" { + continue + } + out[k] = v + } + return out +} + +func (c *itemTraceCtx) setRecipeUpload(up *hyfile.UploadPDFResult, localPath string, err error) { + if c == nil || c.out == nil { + return + } + tr := &RecipeFileUploadTrace{Executed: true, LocalPDFPath: localPath} + if up != nil { + tr.FileID = up.FileID + tr.HTTPStatus = up.HTTPStatus + tr.RawBody = up.RawBody + tr.Success = up.Success + tr.Message = up.Message + } + if err != nil { + tr.Error = err.Error() + } + c.out.RecipeFileUpload = tr +} + +func (c *itemTraceCtx) setRecipeSkipped(reason string) { + if c == nil || c.out == nil { + return + } + c.out.RecipeFileUpload = &RecipeFileUploadTrace{ + Skipped: true, + SkipReason: reason, + } +} + +func (c *itemTraceCtx) finish(cb xkapi.CallbackItem) { + if c == nil || c.out == nil { + return + } + c.out.Callback = cb + c.out.Push = PushTrace{ + HTTPCode: cb.HTTPCode, + MsgCode: cb.MsgCode, + Msg: cb.Msg, + TraceID: cb.TraceID, + ResponseBody: cb.ResponseBody, + } +} diff --git a/internal/sync/step_test_run_test.go b/internal/sync/step_test_run_test.go new file mode 100644 index 0000000..2279372 --- /dev/null +++ b/internal/sync/step_test_run_test.go @@ -0,0 +1,14 @@ +package syncer + +import ( + "strings" + "testing" +) + +func TestTestSuperviseStep_rejectsOtherSteps(t *testing.T) { + r := &Runner{} + _, err := r.TestSuperviseStep("referral", "2026-05-18", 1) + if err == nil || !strings.Contains(err.Error(), "recipe or verification") { + t.Fatalf("err=%v", err) + } +} diff --git a/internal/sync/supervise_step.go b/internal/sync/supervise_step.go new file mode 100644 index 0000000..b9be138 --- /dev/null +++ b/internal/sync/supervise_step.go @@ -0,0 +1,23 @@ +package syncer + +import ( + "fmt" + "strings" +) + +// TestSuperviseStep 联调：仅 recipe / verification，与自动 sync 共用 executeStep。 +func (r *Runner) TestSuperviseStep(step, anchorDate string, limit int) (*StepTestResult, error) { + step = strings.TrimSpace(step) + switch step { + case "recipe", "verification": + default: + return nil, fmt.Errorf("step must be recipe or verification, got %q", step) + } + if limit < 0 { + limit = 0 + } + return r.executeStep(step, anchorDate, StepRunOptions{ + ItemLimit: limit, + CollectTrace: true, + }) +} diff --git a/internal/sync/sync.go b/internal/sync/sync.go index 0ec2ea2..0db1e2a 100644 --- a/internal/sync/sync.go +++ b/internal/sync/sync.go @@ -117,30 +117,55 @@ func (r *Runner) Run(step, anchorDate string) error { // 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 fmt.Errorf("unknown step: %s", step) + 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) - _, _ = r.store.UpsertJob(anchorDate, step, "running", 0, 0, 0, "") + if r.store != nil { + _, _ = 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") + 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 fmt.Errorf("batch create: %w", 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) - // ② 云端组包拉取（含 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") + 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 err + return result, err } validationCount := 0 @@ -150,14 +175,37 @@ func (r *Runner) runStep(step, anchorDate string) error { } } applog.Pullf("pull done step=%s batch_id=%d items=%d validation_failed=%d", step, batchID, len(pull.Items), validationCount) + 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 - - // ③ 逐条：recipe 可能先 PDF 上传，再加密经 forward 报政务云 - for _, item := range pull.Items { - cb := r.processItem(step, anchorDate, method, batchID, item) + for _, item := range items { + var tr *itemTraceCtx + if opts.CollectTrace { + tr = newItemTraceCtx(item) + if item.BizKey == "" && item.Payload != nil { + tr.out.BizKey = hy.BizKey(step, anchorDate, item.Payload) + } + } + cb := r.processItem(step, anchorDate, method, 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++ @@ -168,22 +216,35 @@ func (r *Runner) runStep(step, anchorDate string) error { } } - // ④ 整批回调云端（写 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) + 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") + _ = r.xk.BatchFinish(batchID, "done") total := success + failed + skipped - _, _ = r.store.UpsertJob(anchorDate, step, "done", total, success, failed, "") - _ = r.store.FinishJob(anchorDate, step, "done") + 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) - return nil + 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) xkapi.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 { @@ -203,7 +264,9 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item bussID = fmt.Sprintf("%v", payload["bussID"]) } payloadJSON, _ := json.Marshal(payload) - _ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, bussID, bizKey, "", string(payloadJSON), item.ValidationErrors) + 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" @@ -214,39 +277,57 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item 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 + 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) - _ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", string(payloadJSON), nil) + 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 { - _ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error()) + 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 { - _ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error()) + if r.store != nil { + _ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error()) + } cb.PushStatus = "failed" cb.CallbackStatus = "failed" cb.ErrorMessage = err.Error() @@ -254,9 +335,11 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item 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) + 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 @@ -265,13 +348,17 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item cb.ResponseBody = result.Body if hy.IsSuperviseOK(result) { - _ = r.store.UpdateRecordStatus(bizKey, "success", "") + 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) - _ = r.store.UpdateRecordStatus(bizKey, "failed", errText) + if r.store != nil { + _ = r.store.UpdateRecordStatus(bizKey, "failed", errText) + } cb.PushStatus = "failed" cb.CallbackStatus = "failed" cb.ErrorMessage = errText @@ -289,7 +376,7 @@ func ResolveAnchorDate(cfg config.Config, dateArg string) string { } // ensureRecipeFileID 处方 PDF 子流程：HTML→PDF→本地落盘→经 forward 上传→写回 recipeFileId。 -func (r *Runner) ensureRecipeFileID(item xkapi.PullItem, payload map[string]any) error { +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 { @@ -334,11 +421,15 @@ func (r *Runner) ensureRecipeFileID(item xkapi.PullItem, payload map[string]any) r.configMu.RLock() url := r.fileUploadURL r.configMu.RUnlock() - fileID, err := hyfile.UploadPDF(pdf, filename, url, token, r.cfg.ForwardSharedSecret) + 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，并回写云端处方表 @@ -528,9 +619,10 @@ func (r *Runner) RetryRecord(localRecordID int64) error { ValidationErrors: validationErrors, } - cb := r.processItem(step, rec.AnchorDate, method, batchID, item) + 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 warning record_id=%d err=%v", localRecordID, err) + 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" {