IaC / SwiftFlow

Artifact ef4d64a14dfcc530cbad9abe3b4b543f48f778067d466feda9c726b51df56dca:

• File backend/src/sync_inbounds.gleam — part of check-in [74ded990b8] at 2026-10-23 05:24:52 on branch trunk — 总仓的采购入库单改为只读同步，并在本地生成批次码。

  轮询和人工同步都只拉取已记账的总仓采购入库单，红冲和只作显示的仓库不入库。批次码写在本地，不回来肯。 (user: pluto size: 7227)

import app.{type Context}
import clock
import config
import database
import gleam/list
import gleam/option.{Some}
import gleam/result
import ql361api.{type PurchaseOrder, type PurchaseOrderSummary}
import ql361api/purchases as ql_purchases
import repo.{SyncRun}
import session
import sync_lock
import warehouse_scope
import sync_report.{
  type SyncError, type SyncReport, SyncBusy, SyncDb, SyncNotConfigured, SyncQl361,
  SyncRejected, SyncReport,
}
import wisp

const max_pages = 500

pub fn run(ctx: Context, trigger: String) -> Result(SyncReport, SyncError) {
  case ctx.config.ql361_username != "" && ctx.config.ql361_password != "" {
    False -> Error(SyncNotConfigured)
    True ->
      case sync_lock.with_lock(ctx.lock, "inbounds", fn() { do_sync(ctx, trigger) }) {
        Error(Nil) -> Error(SyncBusy)
        Ok(result) -> result
      }
  }
}

fn do_sync(ctx: Context, trigger: String) -> Result(SyncReport, SyncError) {
  let started = clock.now_local()
  let #(begin_date, end_date) = config.order_window(ctx.config)
  let orders_result = collect_pages(ctx, begin_date, end_date, 0, [])
  let warehouses_result = case orders_result {
    Error(_) -> Error(SyncRejected(""))
    Ok(_) ->
      case session.with_client(ctx.session, ql361api.fetch_warehouses) {
        Ok(rows) -> Ok(rows)
        Error(err) -> Error(SyncQl361(err))
      }
  }

  case orders_result, warehouses_result {
    Error(err), _ -> {
      record_failure(ctx, trigger, started, sync_report.describe(err))
      Error(err)
    }
    _, Error(err) -> {
      record_failure(ctx, trigger, started, sync_report.describe(err))
      Error(err)
    }
    Ok(orders), Ok(warehouses) ->
      write_orders(ctx, trigger, started, begin_date, end_date, orders, warehouses)
  }
}

fn write_orders(
  ctx: Context,
  trigger: String,
  started: String,
  begin_date: String,
  end_date: String,
  orders: List(PurchaseOrderSummary),
  warehouses: List(ql361api.QlWarehouse),
) -> Result(SyncReport, SyncError) {
  let tokens = ctx.config.stock_warehouses
  let stock_ids = warehouse_scope.stock_warehouse_ids(warehouses, tokens)
  let orders =
    list.filter(orders, fn(order) {
      warehouse_scope.bill_in_stock_scope(
        order.warehouse_id,
        order.warehouse_name,
        warehouses,
        tokens,
      )
    })
  let prepared =
    database.transaction(ctx.db, fn(conn) {
      use pending <- result.try(repo.list_pending_inbound_bill_ids(
        conn,
        begin_date,
        end_date,
        stock_ids,
      ))
      use pairs <- result.try(list.try_map(orders, fn(order) {
        use stamp <- result.try(repo.get_inbound_stamp(conn, order.bill_id))
        Ok(#(order, stamp))
      }))
      Ok(#(pending, pairs))
    })

  use #(pending, pairs) <- result_db(prepared)
  let need =
    list.filter(pairs, fn(pair) { repo.needs_inbound_detail(pair.1, pair.0) })
    |> list.map(fn(pair) { pair.0.bill_id })
  let seen = list.map(orders, fn(order) { order.bill_id })
  let missing =
    list.filter(pending, fn(bill_id) { !list.contains(seen, bill_id) })
  let #(details, detail_errors) =
    fetch_details(ctx, list.append(need, missing), [], [])

  let finished = clock.now_local()
  let status = case detail_errors {
    [] -> "ok"
    _ -> "error"
  }
  let error_message = case detail_errors {
    [] -> ""
    [err, ..] -> sync_report.describe(err)
  }
  let run =
    SyncRun(
      id: clock.now_local() <> "-" <> wisp.random_string(8),
      entity: "inbounds",
      trigger: trigger,
      started_at: started,
      finished_at: finished,
      status: status,
      fetched: list.length(orders),
      upserted: list.length(orders),
      error_message: error_message,
    )

  let written =
    database.transaction(ctx.db, fn(conn) {
      use _ <- result.try(repo.delete_documents_outside_warehouses(conn, stock_ids))
      use _ <- result.try(list.try_fold(warehouses, Nil, fn(_, warehouse) {
        repo.upsert_warehouse(
          conn,
          warehouse,
          list.contains(stock_ids, warehouse.id),
          finished,
        )
      }))
      use _ <- result.try(list.try_fold(orders, Nil, fn(_, order) {
        repo.upsert_inbound_summary(conn, order, 504, finished)
      }))
      use _ <- result.try(list.try_fold(details, Nil, fn(_, order) {
        repo.save_inbound_detail(conn, order, 504, finished)
      }))
      use _ <- result.try(repo.set_cursor(conn, "inbounds", finished, finished))
      repo.insert_run(conn, run)
    })

  case written, detail_errors {
    Error(err), _ -> Error(SyncDb(err))
    Ok(_), [] ->
      Ok(SyncReport(
        entity: "inbounds",
        trigger: trigger,
        fetched: list.length(orders),
        upserted: list.length(orders),
        cursor: finished,
      ))
    Ok(_), [err, ..] -> Error(err)
  }
}

fn collect_pages(
  ctx: Context,
  begin_date: String,
  end_date: String,
  page: Int,
  acc: List(PurchaseOrderSummary),
) -> Result(List(PurchaseOrderSummary), SyncError) {
  case page >= max_pages {
    True -> Error(SyncRejected("采购订单分页超过 500 页，已中止"))
    False -> {
      let fetched =
        session.with_client(ctx.session, fn(client) {
          ql361api.fetch_purchase_order_page(
            client,
            begin_date,
            end_date,
            Some(ql_purchases.pending_receipt_status),
            page,
            ql_purchases.page_size,
          )
        })
      case fetched {
        Error(err) -> Error(SyncQl361(err))
        Ok(page_data) -> {
          let acc = list.append(acc, page_data.items)
          let count = list.length(page_data.items)
          let finished =
            count == 0
            || count < ql_purchases.page_size
            || list.length(acc) >= page_data.total
          case finished {
            True -> Ok(acc)
            False -> collect_pages(ctx, begin_date, end_date, page + 1, acc)
          }
        }
      }
    }
  }
}

fn fetch_details(
  ctx: Context,
  bill_ids: List(Int),
  ok: List(PurchaseOrder),
  errors: List(SyncError),
) -> #(List(PurchaseOrder), List(SyncError)) {
  case bill_ids {
    [] -> #(list.reverse(ok), list.reverse(errors))
    [bill_id, ..rest] -> {
      let fetched =
        session.with_client(ctx.session, fn(client) {
          ql361api.fetch_purchase_order_detail(client, bill_id)
        })
      case fetched {
        Ok(order) -> fetch_details(ctx, rest, [order, ..ok], errors)
        Error(err) -> fetch_details(ctx, rest, ok, [SyncQl361(err), ..errors])
      }
    }
  }
}

fn record_failure(
  ctx: Context,
  trigger: String,
  started: String,
  message: String,
) -> Nil {
  let run =
    SyncRun(
      id: clock.now_local() <> "-" <> wisp.random_string(8),
      entity: "inbounds",
      trigger: trigger,
      started_at: started,
      finished_at: clock.now_local(),
      status: "error",
      fetched: 0,
      upserted: 0,
      error_message: message,
    )
  let _ = database.transaction(ctx.db, fn(conn) { repo.insert_run(conn, run) })
  Nil
}

fn result_db(
  result: Result(a, database.DbError),
  next: fn(a) -> Result(b, SyncError),
) -> Result(b, SyncError) {
  case result {
    Ok(value) -> next(value)
    Error(err) -> Error(SyncDb(err))
  }
}