667 lines
21 KiB
Go
667 lines
21 KiB
Go
// 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
|
||
}
|