431 lines
14 KiB
Go
431 lines
14 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"
|
||
"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()
|
||
}
|
||
}
|
||
|
||
// 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)
|
||
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)
|
||
_ = r.store.SaveLog(recordLocalID, result.HTTPCode, result.MsgCode, result.Msg, result.TraceID, len(req.Body), 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)
|
||
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 }
|