Compare commits

..
Author SHA1 Message Date
QiuSW 5dcff4b15a feat(admin): authorize batch purchase starts 2026-08-04 18:24:39 +08:00
QiuSW e379d50101 chore(tasks): extend T-203 auth review scope 2026-08-04 18:18:19 +08:00
27 changed files with 1964 additions and 1638 deletions
+7
View File
@@ -9,6 +9,9 @@
| `CMBUYER_SESSION_SECRET` | 至少 32 字节的会话签名密钥。 |
| `CMBUYER_COOKIE_SECURE` | 可选;存在时只能精确为 `true` 或 `false`。HTTPS 部署应设为 `true`。 |
| `CMBUYER_DATABASE_SOURCE` | 已迁移 SQLite 的显式 data source。 |
| `CMBUYER_AUTHORIZATION_TTL` | 一次性授权的正 Go duration,例如 `10m`。 |
| `CMBUYER_MAX_TASK_QUANTITY` | 每条任务允许的正整数数量上限。 |
| `CMBUYER_MAX_TOTAL_PRICE` | 每条任务允许的规范正数总价上限,例如 `999.99`。 |
示例仅展示变量名,不提供可运行凭据:
@@ -18,8 +21,12 @@ $env:CMBUYER_ADMIN_PASSWORD_BCRYPT = '<bcrypt 密码哈希>'
$env:CMBUYER_SESSION_SECRET = '<至少 32 字节的随机密钥>'
$env:CMBUYER_COOKIE_SECURE = 'true'
$env:CMBUYER_DATABASE_SOURCE = '<SQLite data source>'
$env:CMBUYER_AUTHORIZATION_TTL = '10m'
$env:CMBUYER_MAX_TASK_QUANTITY = '99'
$env:CMBUYER_MAX_TOTAL_PRICE = '999.99'
go run ./cmd/migrate -database $env:CMBUYER_DATABASE_SOURCE up
go run ./cmd/server
```
采购服务会话仅保存在当前进程内;进程重启后既有登录会话会安全失效。
管理员的“开始采购(只创建待付款订单)”只签发一次性授权并创建待付款订单的资格;服务不会自动付款,也不包含任何支付操作。
+1
View File
@@ -34,6 +34,7 @@ func run() error {
if err != nil {
return err
}
taskStore.SetStartPolicy(tasks.StartPolicy{AuthorizationTTL: configuration.AuthorizationTTL, MaxQuantity: configuration.MaxTaskQuantity, MaxTotalPrice: configuration.MaxTotalPrice})
router, err := server.NewRouter(server.Options{
AdminUsername: configuration.AdminUsername,
+6
View File
@@ -62,6 +62,12 @@ func (manager *Manager) Ensure(writer http.ResponseWriter, request *http.Request
return current.csrfToken, false
}
// IsAuthenticated 只读检查当前请求是否持有有效管理会话;它不会像 Ensure 一样创建匿名会话。
func (manager *Manager) IsAuthenticated(request *http.Request) bool {
_, current, found := manager.current(request)
return found && current.authenticated
}
// VerifyCSRF 只接受当前未过期会话中以恒定时间比较匹配的 token。
func (manager *Manager) VerifyCSRF(request *http.Request, token string) (authenticated bool, ok bool) {
_, current, found := manager.current(request)
+45
View File
@@ -37,6 +37,51 @@ func TestManagerRejectsTamperedAndExpiredCookies(t *testing.T) {
}
}
func TestIsAuthenticatedDoesNotCreateOrDependOnCSRFValidation(t *testing.T) {
manager := NewManager([]byte(strings.Repeat("s", 32)), false)
missingSession := httptest.NewRequest(http.MethodPost, "/tasks/start-purchases", nil)
if manager.IsAuthenticated(missingSession) {
t.Fatal("missing session was treated as authenticated")
}
if len(manager.sessions) != 0 {
t.Fatalf("read-only authentication check created %d sessions", len(manager.sessions))
}
anonymousRequest := httptest.NewRequest(http.MethodGet, "/login", nil)
anonymousResponse := httptest.NewRecorder()
manager.Ensure(anonymousResponse, anonymousRequest)
anonymousCookie := anonymousResponse.Result().Cookies()[0]
anonymousCheck := httptest.NewRequest(http.MethodPost, "/tasks/start-purchases", nil)
anonymousCheck.AddCookie(anonymousCookie)
if manager.IsAuthenticated(anonymousCheck) {
t.Fatal("anonymous CSRF session was treated as authenticated")
}
loginRequest := httptest.NewRequest(http.MethodPost, "/login", nil)
loginRequest.AddCookie(anonymousCookie)
authenticatedResponse := httptest.NewRecorder()
csrf := manager.RotateAuthenticated(authenticatedResponse, loginRequest)
authenticatedCookie := authenticatedResponse.Result().Cookies()[0]
authenticatedCheck := httptest.NewRequest(http.MethodPost, "/tasks/start-purchases", nil)
authenticatedCheck.AddCookie(authenticatedCookie)
if !manager.IsAuthenticated(authenticatedCheck) {
t.Fatal("valid authenticated session was not recognized")
}
if authenticated, csrfOK := manager.VerifyCSRF(authenticatedCheck, "wrong-token"); authenticated || csrfOK {
t.Fatalf("wrong token result = (%t, %t), want (false, false)", authenticated, csrfOK)
}
validRequest := httptest.NewRequest(http.MethodPost, "/tasks/start-purchases", nil)
validRequest.AddCookie(authenticatedCookie)
if authenticated, csrfOK := manager.VerifyCSRF(validRequest, csrf); !authenticated || !csrfOK {
t.Fatalf("valid token result = (%t, %t), want (true, true)", authenticated, csrfOK)
}
if authenticated, csrfOK := manager.VerifyCSRF(httptest.NewRequest(http.MethodPost, "/tasks/start-purchases", nil), csrf); authenticated || csrfOK {
t.Fatalf("missing session result = (%t, %t), want (false, false)", authenticated, csrfOK)
}
}
func flipCookieValue(t *testing.T, value string) string {
t.Helper()
if value == "" {
+47
View File
@@ -5,7 +5,9 @@ import (
"errors"
"fmt"
"os"
"strconv"
"strings"
"time"
"golang.org/x/crypto/bcrypt"
)
@@ -16,6 +18,9 @@ const (
sessionSecretEnv = "CMBUYER_SESSION_SECRET"
cookieSecureEnv = "CMBUYER_COOKIE_SECURE"
databaseSourceEnv = "CMBUYER_DATABASE_SOURCE"
authorizationTTLEnv = "CMBUYER_AUTHORIZATION_TTL"
maxTaskQuantityEnv = "CMBUYER_MAX_TASK_QUANTITY"
maxTotalPriceEnv = "CMBUYER_MAX_TOTAL_PRICE"
minimumSecretLength = 32
)
@@ -26,6 +31,9 @@ type Config struct {
SessionSecret []byte
CookieSecure bool
DatabaseSource string
AuthorizationTTL time.Duration
MaxTaskQuantity int
MaxTotalPrice string
}
// LoadFromEnv 从进程环境读取配置。错误只指出缺失或非法的变量名,绝不回显秘密。
@@ -71,6 +79,29 @@ func Load(lookup func(string) (string, bool)) (Config, error) {
if err != nil {
return Config{}, err
}
ttlText, err := required(lookup, authorizationTTLEnv)
if err != nil {
return Config{}, err
}
ttl, err := time.ParseDuration(ttlText)
if err != nil || ttl <= 0 {
return Config{}, fmt.Errorf("%s must be a positive duration", authorizationTTLEnv)
}
quantityText, err := required(lookup, maxTaskQuantityEnv)
if err != nil {
return Config{}, err
}
maxQuantity, err := strconv.Atoi(quantityText)
if err != nil || maxQuantity < 1 {
return Config{}, fmt.Errorf("%s must be a positive integer", maxTaskQuantityEnv)
}
maxPrice, err := required(lookup, maxTotalPriceEnv)
if err != nil {
return Config{}, err
}
if !canonicalMoney(maxPrice) {
return Config{}, fmt.Errorf("%s must be a canonical positive decimal", maxTotalPriceEnv)
}
return Config{
AdminUsername: username,
@@ -78,9 +109,25 @@ func Load(lookup func(string) (string, bool)) (Config, error) {
SessionSecret: []byte(secret),
CookieSecure: cookieSecure,
DatabaseSource: databaseSource,
AuthorizationTTL: ttl, MaxTaskQuantity: maxQuantity, MaxTotalPrice: maxPrice,
}, nil
}
func canonicalMoney(value string) bool {
parts := strings.Split(value, ".")
if len(parts) != 2 || len(parts[0]) == 0 || len(parts[1]) != 2 || (len(parts[0]) > 1 && parts[0][0] == '0') {
return false
}
for _, part := range parts {
for _, ch := range part {
if ch < '0' || ch > '9' {
return false
}
}
}
return strings.Trim(parts[0]+parts[1], "0") != ""
}
func required(lookup func(string) (string, bool), name string) (string, error) {
value, present := lookup(name)
if !present || strings.TrimSpace(value) == "" {
+9
View File
@@ -21,6 +21,9 @@ func TestLoad(t *testing.T) {
"CMBUYER_SESSION_SECRET": strings.Repeat("s", 32),
"CMBUYER_COOKIE_SECURE": "true",
"CMBUYER_DATABASE_SOURCE": ":memory:",
"CMBUYER_AUTHORIZATION_TTL": "10m",
"CMBUYER_MAX_TASK_QUANTITY": "99",
"CMBUYER_MAX_TOTAL_PRICE": "999.99",
}
got, err := config.Load(lookup(values))
@@ -43,6 +46,9 @@ func TestLoadRejectsMissingOrInvalidConfiguration(t *testing.T) {
"CMBUYER_ADMIN_PASSWORD_BCRYPT": string(hash),
"CMBUYER_SESSION_SECRET": strings.Repeat("s", 32),
"CMBUYER_DATABASE_SOURCE": ":memory:",
"CMBUYER_AUTHORIZATION_TTL": "10m",
"CMBUYER_MAX_TASK_QUANTITY": "99",
"CMBUYER_MAX_TOTAL_PRICE": "999.99",
}
tests := []struct {
@@ -55,6 +61,9 @@ func TestLoadRejectsMissingOrInvalidConfiguration(t *testing.T) {
{"short secret", func(values map[string]string) { values["CMBUYER_SESSION_SECRET"] = "short" }, "CMBUYER_SESSION_SECRET"},
{"invalid secure flag", func(values map[string]string) { values["CMBUYER_COOKIE_SECURE"] = "1" }, "CMBUYER_COOKIE_SECURE"},
{"missing database", func(values map[string]string) { delete(values, "CMBUYER_DATABASE_SOURCE") }, "CMBUYER_DATABASE_SOURCE"},
{"invalid authorization ttl", func(values map[string]string) { values["CMBUYER_AUTHORIZATION_TTL"] = "0s" }, "CMBUYER_AUTHORIZATION_TTL"},
{"invalid maximum quantity", func(values map[string]string) { values["CMBUYER_MAX_TASK_QUANTITY"] = "0" }, "CMBUYER_MAX_TASK_QUANTITY"},
{"invalid maximum total price", func(values map[string]string) { values["CMBUYER_MAX_TOTAL_PRICE"] = "1" }, "CMBUYER_MAX_TOTAL_PRICE"},
}
for _, test := range tests {
+142 -13
View File
@@ -2,11 +2,16 @@
package server
import (
"bytes"
"crypto/subtle"
"encoding/json"
"errors"
"io"
"mime"
"net/http"
"net/url"
"strings"
"unicode/utf8"
"cmbuyer/admin/internal/auth"
"cmbuyer/admin/internal/tasks"
@@ -17,6 +22,7 @@ import (
)
const maxFormBytes = 8 << 10
const maxJSONBytes = 64 << 10
// Options 是路由层需要的安全依赖。凭据由启动配置注入,不能在路由中设置默认值。
type Options struct {
@@ -42,10 +48,84 @@ func NewRouter(options Options) (*gin.Engine, error) {
router.GET("/tasks", tasksPage(options))
router.GET("/tasks/new", newTaskPage(options))
router.POST("/tasks", createTask(options))
router.POST("/tasks/start-purchases", startPurchases(options))
router.GET("/static/tasks.js", func(context *gin.Context) {
context.Data(http.StatusOK, "application/javascript; charset=utf-8", webui.TasksScript())
})
return router, nil
}
func startPurchases(options Options) gin.HandlerFunc {
return func(context *gin.Context) {
if !options.Sessions.IsAuthenticated(context.Request) {
context.Status(http.StatusUnauthorized)
return
}
authenticated, csrfOK := options.Sessions.VerifyCSRF(context.Request, context.GetHeader("X-CSRF-Token"))
if !authenticated || !csrfOK {
context.Status(http.StatusForbidden)
return
}
if !isJSONContentType(context.GetHeader("Content-Type")) {
context.Status(http.StatusUnsupportedMediaType)
return
}
context.Request.Body = http.MaxBytesReader(context.Writer, context.Request.Body, maxJSONBytes)
raw, err := io.ReadAll(context.Request.Body)
if err != nil {
var tooLarge *http.MaxBytesError
if errors.As(err, &tooLarge) {
context.Status(http.StatusRequestEntityTooLarge)
} else {
context.Status(http.StatusBadRequest)
}
return
}
if !utf8.Valid(raw) {
context.Status(http.StatusBadRequest)
return
}
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.DisallowUnknownFields()
var command tasks.StartCommand
if err := decoder.Decode(&command); err != nil {
context.Status(http.StatusBadRequest)
return
}
var extra any
if err := decoder.Decode(&extra); err != io.EOF {
context.Status(http.StatusBadRequest)
return
}
result, err := options.Tasks.StartPurchases(context.Request.Context(), command, options.AdminUsername)
if err != nil {
if errors.Is(err, tasks.ErrInvalidStart) {
context.Status(http.StatusBadRequest)
} else if errors.Is(err, tasks.ErrStartConflict) {
context.Status(http.StatusConflict)
} else {
context.Status(http.StatusInternalServerError)
}
return
}
context.JSON(http.StatusOK, result)
}
}
func isJSONContentType(value string) bool {
mediaType, parameters, err := mime.ParseMediaType(value)
if err != nil || mediaType != "application/json" {
return false
}
for name, value := range parameters {
if name != "charset" || !strings.EqualFold(value, "utf-8") {
return false
}
}
return true
}
func healthz(context *gin.Context) {
context.JSON(http.StatusOK, gin.H{"status": "ok"})
}
@@ -55,7 +135,7 @@ func securityHeaders() gin.HandlerFunc {
context.Header("Cache-Control", "no-store")
context.Header("X-Content-Type-Options", "nosniff")
context.Header("Referrer-Policy", "no-referrer")
context.Header("Content-Security-Policy", "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'none'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'")
context.Header("Content-Security-Policy", "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'")
context.Next()
}
}
@@ -126,14 +206,23 @@ func tasksPage(options Options) gin.HandlerFunc {
return
}
drafts, err := options.Tasks.ListDrafts(context.Request.Context())
filter := tasks.TaskFilter{Keyword: context.Query("keyword"), Status: context.Query("status"), CreatedFrom: context.Query("created_from"), CreatedTo: context.Query("created_to")}
if validation := tasks.ValidateTaskFilter(filter); !validation.Valid() {
startKey, err := tasks.NewCreateKey()
if err != nil {
context.Status(http.StatusInternalServerError)
return
}
renderTasks(context, http.StatusBadRequest, webui.TasksData{CSRFToken: csrfToken, Filter: filter, FilterErrors: validation, HasFilter: true, StartKey: startKey})
return
}
data, err := taskListData(context, options, csrfToken, filter)
if err != nil {
context.Status(http.StatusInternalServerError)
return
}
data := webui.TasksData{CSRFToken: csrfToken, Drafts: drafts}
for _, draft := range drafts {
if draft.ID == context.Query("created") {
for _, row := range data.Tasks {
if row.ID == context.Query("created") {
data.Success = true
break
}
@@ -185,24 +274,32 @@ func createTask(options Options) gin.HandlerFunc {
}
fullPage := requestForm.Get("form_mode") == "full"
if !validation.Valid() {
drafts, err := options.Tasks.ListDrafts(context.Request.Context())
if err != nil {
context.Status(http.StatusInternalServerError)
data, ok := createErrorData(context, options, fullPage)
if !ok {
return
}
renderTasks(context, http.StatusBadRequest, webui.TasksData{CSRFToken: csrfFor(context, options), Drafts: drafts, Form: form, Errors: validation, OpenForm: !fullPage, FullPage: fullPage, FocusField: firstError(validation)})
data.Form = form
data.Errors = validation
data.OpenForm = !fullPage
data.FullPage = fullPage
data.FocusField = firstError(validation)
renderTasks(context, http.StatusBadRequest, data)
return
}
created, err := options.Tasks.CreateDraft(context.Request.Context(), draft)
if err != nil {
if errors.Is(err, tasks.ErrCreateKeyConflict) {
validation["create_key"] = "该创建请求已用于另一条任务,请重新打开表单。"
drafts, listErr := options.Tasks.ListDrafts(context.Request.Context())
if listErr != nil {
context.Status(http.StatusInternalServerError)
data, ok := createErrorData(context, options, fullPage)
if !ok {
return
}
renderTasks(context, http.StatusConflict, webui.TasksData{CSRFToken: csrfFor(context, options), Drafts: drafts, Form: form, Errors: validation, OpenForm: !fullPage, FullPage: fullPage, FocusField: firstError(validation)})
data.Form = form
data.Errors = validation
data.OpenForm = !fullPage
data.FullPage = fullPage
data.FocusField = firstError(validation)
renderTasks(context, http.StatusConflict, data)
return
}
context.Status(http.StatusInternalServerError)
@@ -226,6 +323,38 @@ func csrfFor(context *gin.Context, options Options) string {
csrf, _ := options.Sessions.Ensure(context.Writer, context.Request)
return csrf
}
func taskListData(context *gin.Context, options Options, csrfToken string, filter tasks.TaskFilter) (webui.TasksData, error) {
rows, err := options.Tasks.ListTasks(context.Request.Context(), filter)
if err != nil {
return webui.TasksData{}, err
}
startKey, err := tasks.NewCreateKey()
if err != nil {
return webui.TasksData{}, err
}
return webui.TasksData{
CSRFToken: csrfToken,
Tasks: rows,
Filter: filter,
HasFilter: filter.Keyword != "" || filter.Status != "" || filter.CreatedFrom != "" || filter.CreatedTo != "",
StartKey: startKey,
}, nil
}
func createErrorData(context *gin.Context, options Options, fullPage bool) (webui.TasksData, bool) {
csrfToken := csrfFor(context, options)
if fullPage {
return webui.TasksData{CSRFToken: csrfToken}, true
}
data, err := taskListData(context, options, csrfToken, tasks.TaskFilter{})
if err != nil {
context.Status(http.StatusInternalServerError)
return webui.TasksData{}, false
}
return data, true
}
func renderTasks(context *gin.Context, status int, data webui.TasksData) {
context.Header("Content-Type", "text/html; charset=utf-8")
context.Status(status)
+125 -5
View File
@@ -8,6 +8,7 @@ import (
"regexp"
"strings"
"testing"
"time"
"cmbuyer/admin/internal/auth"
"cmbuyer/admin/internal/server"
@@ -190,7 +191,7 @@ func TestTaskCreationRendersSharedFormsAndPersistsOnlyDraft(t *testing.T) {
if fullPage.Code != http.StatusOK {
t.Fatalf("GET full form status = %d, want 200", fullPage.Code)
}
for _, want := range []string{`<div class="modal-scrim"`, `<dialog open`, `aria-modal="true"`, `name="title"`, `name="product_url"`, `name="sku_color"`, `name="sku_size"`, `name="quantity"`, `name="max_total_price"`, `type="url" inputmode="url" maxlength="2048"`, `type="number" inputmode="numeric" min="1" step="1"`, `inputmode="decimal" pattern="[0-9]+(\.[0-9]{1,2})?"`, `maxlength="120"`, `maxlength="80"`, `required`, `autofocus`, `导入</button><a class="button primary"`, `type="search" disabled`, `disabled>筛选</button>`, `disabled>清除</button>`, `min-height:44px`, `overflow-x:auto`, `prefers-reduced-motion`} {
for _, want := range []string{`<div class="modal-scrim"`, `<dialog open`, `aria-modal="true"`, `name="title"`, `name="product_url"`, `name="sku_color"`, `name="sku_size"`, `name="quantity"`, `name="max_total_price"`, `type="url" inputmode="url" maxlength="2048"`, `type="number" inputmode="numeric" min="1" step="1"`, `inputmode="decimal" pattern="[0-9]+(\.[0-9]{1,2})?"`, `maxlength="120"`, `maxlength="80"`, `required`, `autofocus`, `导入</button><a class="button primary"`, `type="search"`, `data-start-purchases`, `data-select-all`, `最高总额`, `min-height:44px`, `:focus-visible`, `overflow-x:auto`, `prefers-reduced-motion`} {
if !strings.Contains(modal.Body.String(), want) {
t.Fatalf("dialog form is missing %q", want)
}
@@ -277,13 +278,108 @@ func TestTaskCreationRendersSharedFormsAndPersistsOnlyDraft(t *testing.T) {
t.Fatalf("task list is missing %q", want)
}
}
for _, forbidden := range []string{"utm_source", "试选", "PENDING", "支付", "订单确认", "真机", "提交订单"} {
for _, forbidden := range []string{"utm_source", "试选", "订单确认", "真机", "提交订单"} {
if strings.Contains(body, forbidden) {
t.Fatalf("task list exposed deferred scope %q", forbidden)
}
}
}
func TestTasksPageKeepsOriginalShellAndRendersFilteredWorkbench(t *testing.T) {
store := &memoryStore{rows: []tasks.TaskRow{
{ID: "b3c9f507-7473-4fa6-8d71-8786c34c6301", Title: "待开始衬衫", GoodsID: "937122477375", SKUColor: "黑色", SKUSize: "M", Quantity: 2, MaxTotalPrice: "12.80", Status: "DRAFT", Version: 3, CreatedAt: time.Date(2026, 8, 4, 1, 2, 3, 0, time.UTC)},
{ID: "c3c9f507-7473-4fa6-8d71-8786c34c6301", Title: "等待领取衬衫", GoodsID: "958756616606", SKUColor: "白色", SKUSize: "L", Quantity: 1, MaxTotalPrice: "20.00", Status: "PENDING", Version: 4, CreatedAt: time.Date(2026, 8, 4, 2, 3, 4, 0, time.UTC)},
}}
router, _ := newRouterWithStore(t, store)
cookie := authenticate(t, router)
query := url.Values{"keyword": {"衬衫"}, "created_from": {"2026-08-04"}, "created_to": {"2026-08-04"}}
response := serve(router, http.MethodGet, "/tasks?"+query.Encode(), nil, cookie)
if response.Code != http.StatusOK {
t.Fatalf("filtered tasks status = %d, want 200", response.Code)
}
body := response.Body.String()
for _, want := range []string{
`<a class="skip" href="#main">`,
`:focus-visible`,
`min-height:44px`,
`@media(max-width:420px)`,
`prefers-reduced-motion`,
`<button class="button" type="button" disabled>导入</button><a class="button primary" href="/tasks?create=1">创建任务</a>`,
`name="keyword" type="search" value="衬衫"`,
`name="created_from" type="date" value="2026-08-04"`,
`name="created_to" type="date" value="2026-08-04"`,
`data-start-purchases`,
`data-selection-summary aria-live="polite"`,
`系统不会付款`,
`开始采购(只创建待付款订单)`,
`采购结果`,
`创建时间(上海)`,
`https://mobile.yangkeduo.com/goods.html?goods_id=937122477375`,
`target="_blank" rel="noopener noreferrer"`,
`待开始`,
`已授权待领取`,
`datetime="2026-08-04T09:02:03&#43;08:00">2026-08-04 09:02`,
`<script src="/static/tasks.js" defer></script>`,
} {
if !strings.Contains(body, want) {
t.Fatalf("workbench is missing %q", want)
}
}
if strings.Index(body, `name="keyword"`) > strings.Index(body, `data-start-purchases`) || strings.Index(body, `data-start-purchases`) > strings.Index(body, `<div class="table-wrap">`) {
t.Fatal("workbench rows are not ordered as toolbar, filters, batch actions, table")
}
if count := strings.Count(body, `data-task-id=`); count != 1 {
t.Fatalf("selectable row count = %d, want only the DRAFT row", count)
}
for _, forbidden := range []string{`<th scope="col">操作</th>`, `确认开始采购`, `确认机器选对了吗`} {
if strings.Contains(body, forbidden) {
t.Fatalf("workbench exposed forbidden per-row or confirmation UI %q", forbidden)
}
}
if store.listTasksCalls != 1 || store.listDraftsCalls != 0 {
t.Fatalf("GET /tasks calls = (ListTasks %d, ListDrafts %d), want (1, 0)", store.listTasksCalls, store.listDraftsCalls)
}
}
func TestTasksPageRerendersAccessibleFilterErrorsAndKeepsValues(t *testing.T) {
store := &memoryStore{}
router, _ := newRouterWithStore(t, store)
cookie := authenticate(t, router)
query := url.Values{
"keyword": {`保留%_\`},
"status": {"UNKNOWN"},
"created_from": {"2026-02-30"},
"created_to": {"not-a-date"},
}
response := serve(router, http.MethodGet, "/tasks?"+query.Encode(), nil, cookie)
if response.Code != http.StatusBadRequest {
t.Fatalf("invalid filter status = %d, want 400", response.Code)
}
body := response.Body.String()
for _, want := range []string{
`role="alert" aria-live="assertive"`,
`href="#filter-status"`,
`href="#filter-created-from"`,
`href="#filter-created-to"`,
`name="keyword" type="search" value="保留%_\"`,
`<option value="UNKNOWN" selected>无效状态:UNKNOWN</option>`,
`name="created_from" type="date" value="2026-02-30" aria-invalid="true" aria-describedby="filter-created-from-error"`,
`name="created_to" type="date" value="not-a-date" aria-invalid="true" aria-describedby="filter-created-to-error"`,
`id="filter-status-error"`,
`id="filter-created-from-error"`,
`id="filter-created-to-error"`,
`筛选条件有误`,
} {
if !strings.Contains(body, want) {
t.Fatalf("invalid filter page is missing %q", want)
}
}
if store.listTasksCalls != 0 || store.listDraftsCalls != 0 {
t.Fatalf("invalid filter queried stores: ListTasks=%d ListDrafts=%d", store.listTasksCalls, store.listDraftsCalls)
}
assertSecurityHeaders(t, response)
}
func TestTaskCreationRequiresAuthenticationAndCSRF(t *testing.T) {
router, _ := newRouter(t)
if response := serve(router, http.MethodPost, "/tasks", url.Values{}, nil); response.Code != http.StatusForbidden {
@@ -327,7 +423,7 @@ func assertSecurityHeaders(t *testing.T, response *httptest.ResponseRecorder) {
"Cache-Control": "no-store",
"X-Content-Type-Options": "nosniff",
"Referrer-Policy": "no-referrer",
"Content-Security-Policy": "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'none'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'",
"Content-Security-Policy": "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'",
}
for name, expected := range want {
if got := response.Header().Get(name); got != expected {
@@ -381,6 +477,10 @@ func TestLogoutRequiresCSRFAndRevokesSession(t *testing.T) {
}
func newRouter(t *testing.T) (*gin.Engine, *auth.Manager) {
return newRouterWithStore(t, &memoryStore{})
}
func newRouterWithStore(t *testing.T, store tasks.Store) (*gin.Engine, *auth.Manager) {
t.Helper()
gin.SetMode(gin.TestMode)
hash, err := bcrypt.GenerateFromPassword([]byte("test-password"), bcrypt.MinCost)
@@ -392,7 +492,7 @@ func newRouter(t *testing.T) (*gin.Engine, *auth.Manager) {
AdminUsername: "admin",
AdminPasswordBcrypt: string(hash),
Sessions: manager,
Tasks: &memoryStore{},
Tasks: store,
})
if err != nil {
t.Fatalf("NewRouter: %v", err)
@@ -400,7 +500,12 @@ func newRouter(t *testing.T) (*gin.Engine, *auth.Manager) {
return router, manager
}
type memoryStore struct{ drafts []tasks.Draft }
type memoryStore struct {
drafts []tasks.Draft
rows []tasks.TaskRow
listDraftsCalls int
listTasksCalls int
}
func (store *memoryStore) CreateDraft(_ context.Context, draft tasks.Draft) (tasks.Draft, error) {
for _, existing := range store.drafts {
@@ -415,8 +520,23 @@ func (store *memoryStore) CreateDraft(_ context.Context, draft tasks.Draft) (tas
return draft, nil
}
func (store *memoryStore) ListDrafts(_ context.Context) ([]tasks.Draft, error) {
store.listDraftsCalls++
return append([]tasks.Draft(nil), store.drafts...), nil
}
func (store *memoryStore) ListTasks(_ context.Context, _ tasks.TaskFilter) ([]tasks.TaskRow, error) {
store.listTasksCalls++
if store.rows != nil {
return append([]tasks.TaskRow(nil), store.rows...), nil
}
result := make([]tasks.TaskRow, 0, len(store.drafts))
for _, draft := range store.drafts {
result = append(result, tasks.TaskRow{ID: draft.ID, Title: draft.Title, GoodsID: draft.GoodsID, SKUColor: draft.SKUColor, SKUSize: draft.SKUSize, Quantity: draft.Quantity, MaxTotalPrice: draft.MaxTotalPrice, Status: "DRAFT", Version: 1, CreatedAt: draft.CreatedAt})
}
return result, nil
}
func (store *memoryStore) StartPurchases(_ context.Context, _ tasks.StartCommand, _ string) (tasks.StartResult, error) {
return tasks.StartResult{}, tasks.ErrInvalidStart
}
func serve(router http.Handler, method, target string, form url.Values, cookie *http.Cookie) *httptest.ResponseRecorder {
var body *strings.Reader
@@ -0,0 +1,328 @@
package server_test
import (
"bytes"
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"cmbuyer/admin/internal/tasks"
)
const (
startKeyForHTTP = "c3c9f507-7473-4fa6-8d71-8786c34c6301"
taskIDForHTTP = "a3c9f507-7473-4fa6-8d71-8786c34c6301"
)
func TestStartPurchasesAuthenticatesBeforeInspectingRequestBody(t *testing.T) {
store := &startRecordingStore{}
router, _ := newRouterWithStore(t, store)
hugeMalformed := `{"start_key":"` + strings.Repeat("x", 70<<10)
for name, request := range map[string]*http.Request{
"anonymous malformed": newStartRequest(t, hugeMalformed, "text/plain", "", nil),
"device bearer": newStartRequest(t, validStartBody(), "application/json", "", nil),
} {
t.Run(name, func(t *testing.T) {
if name == "device bearer" {
request.Header.Set("Authorization", "Bearer device-token")
}
response := httptest.NewRecorder()
router.ServeHTTP(response, request)
if response.Code != http.StatusUnauthorized {
t.Fatalf("status = %d, want 401", response.Code)
}
})
}
cookie, csrf := authenticatedStartSession(t, router)
for name, token := range map[string]string{"missing CSRF": "", "wrong CSRF": "wrong-csrf"} {
request := newStartRequest(t, hugeMalformed, "text/plain", token, cookie)
response := httptest.NewRecorder()
router.ServeHTTP(response, request)
if response.Code != http.StatusForbidden {
t.Fatalf("%s status = %d, want 403", name, response.Code)
}
}
if csrf == "" {
t.Fatal("authenticated page did not contain a CSRF token")
}
if store.startCalls != 0 {
t.Fatalf("unauthorized requests called store %d times", store.startCalls)
}
}
func TestStartPurchasesRejectsInvalidUTF8BeforeJSONDecoding(t *testing.T) {
validPrefix := []byte(`{"start_key":"` + startKeyForHTTP + `","tasks":[],"start_key":"`)
duplicateKeyBypass := append(append([]byte(nil), validPrefix...), 0xff)
duplicateKeyBypass = append(duplicateKeyBypass, []byte(`"}`)...)
invalidWhitespace := append([]byte(validStartBody()), 0xfe)
for name, body := range map[string][]byte{
"invalid byte after JSON": invalidWhitespace,
"invalid duplicate-key value": duplicateKeyBypass,
} {
t.Run(name, func(t *testing.T) {
store := &startRecordingStore{}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
response := serveStartBytes(t, router, body, "application/json", csrf, cookie)
if response.Code != http.StatusBadRequest || store.startCalls != 0 {
t.Fatalf("status/calls = %d/%d, want 400/0", response.Code, store.startCalls)
}
if response.Body.Len() != 0 {
t.Fatalf("invalid UTF-8 response leaked body %q", response.Body.String())
}
})
}
}
func TestStartPurchasesEnforcesExact64KiBBodyBoundary(t *testing.T) {
const limit = 64 << 10
base := validStartBody()
for name, test := range map[string]struct {
body string
want int
wantCalls int
}{
"exact limit": {body: base + strings.Repeat(" ", limit-len(base)), want: http.StatusOK, wantCalls: 1},
"one over": {body: base + strings.Repeat(" ", limit-len(base)+1), want: http.StatusRequestEntityTooLarge},
} {
t.Run(name, func(t *testing.T) {
store := &startRecordingStore{startResult: successfulStartResult()}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
response := serveStartRequest(t, router, test.body, "application/json", csrf, cookie)
if response.Code != test.want || store.startCalls != test.wantCalls {
t.Fatalf("status/calls = %d/%d, want %d/%d", response.Code, store.startCalls, test.want, test.wantCalls)
}
})
}
}
func TestStartPurchasesContentTypeContract(t *testing.T) {
for _, contentType := range []string{
"application/json",
"application/json; charset=utf-8",
"application/json;charset=UTF-8",
} {
t.Run("accept "+contentType, func(t *testing.T) {
store := &startRecordingStore{startResult: successfulStartResult()}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
response := serveStartRequest(t, router, validStartBody(), contentType, csrf, cookie)
if response.Code != http.StatusOK || store.startCalls != 1 {
t.Fatalf("status/calls = %d/%d, want 200/1", response.Code, store.startCalls)
}
})
}
for _, contentType := range []string{
"",
"text/plain",
"application/json-patch+json",
"application/json; charset=gbk",
"application/json; profile=unapproved",
"application/json; charset",
} {
t.Run("reject "+contentType, func(t *testing.T) {
store := &startRecordingStore{}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
response := serveStartRequest(t, router, validStartBody(), contentType, csrf, cookie)
if response.Code != http.StatusUnsupportedMediaType || store.startCalls != 0 {
t.Fatalf("status/calls = %d/%d, want 415/0", response.Code, store.startCalls)
}
})
}
}
func TestStartPurchasesRejectsMalformedAndOversizedJSON(t *testing.T) {
tests := []struct {
name string
body string
want int
storeErr error
wantCalls int
}{
{name: "empty", body: "", want: http.StatusBadRequest},
{name: "empty object", body: `{}`, want: http.StatusBadRequest, storeErr: tasks.ErrInvalidStart, wantCalls: 1},
{name: "null object", body: `null`, want: http.StatusBadRequest, storeErr: tasks.ErrInvalidStart, wantCalls: 1},
{name: "malformed", body: `{`, want: http.StatusBadRequest},
{name: "wrong top-level type", body: `[]`, want: http.StatusBadRequest},
{name: "unknown field", body: `{"start_key":"` + startKeyForHTTP + `","tasks":[],"created_by":"attacker"}`, want: http.StatusBadRequest},
{name: "wrong field type", body: `{"start_key":"` + startKeyForHTTP + `","tasks":[{"task_id":"` + taskIDForHTTP + `","expected_task_version":"1"}]}`, want: http.StatusBadRequest},
{name: "second JSON value", body: validStartBody() + `{}`, want: http.StatusBadRequest},
{name: "duplicate task ids", body: `{"start_key":"` + startKeyForHTTP + `","tasks":[{"task_id":"` + taskIDForHTTP + `","expected_task_version":1},{"task_id":"` + taskIDForHTTP + `","expected_task_version":1}]}`, want: http.StatusBadRequest, storeErr: tasks.ErrInvalidStart, wantCalls: 1},
{name: "oversized first value", body: `{"start_key":"` + strings.Repeat("x", 70<<10), want: http.StatusRequestEntityTooLarge},
{name: "oversized trailing whitespace", body: validStartBody() + strings.Repeat(" ", 70<<10), want: http.StatusRequestEntityTooLarge},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
store := &startRecordingStore{startErr: test.storeErr}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
response := serveStartRequest(t, router, test.body, "application/json", csrf, cookie)
if response.Code != test.want || store.startCalls != test.wantCalls {
t.Fatalf("status/calls = %d/%d, want %d/%d", response.Code, store.startCalls, test.want, test.wantCalls)
}
if response.Body.Len() != 0 {
t.Fatalf("error response leaked body %q", response.Body.String())
}
assertSecurityHeaders(t, response)
})
}
}
func TestStartPurchasesUsesAuthenticatedAdminAndReturnsStableSafeResult(t *testing.T) {
result := successfulStartResult()
store := &startRecordingStore{startResult: result}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
first := serveStartRequest(t, router, validStartBody(), "application/json; charset=utf-8", csrf, cookie)
second := serveStartRequest(t, router, validStartBody(), "application/json", csrf, cookie)
for index, response := range []*httptest.ResponseRecorder{first, second} {
if response.Code != http.StatusOK {
t.Fatalf("response %d status = %d, want 200", index, response.Code)
}
if got := response.Header().Get("Content-Type"); got != "application/json; charset=utf-8" {
t.Fatalf("response content type = %q", got)
}
var decoded tasks.StartResult
if err := json.Unmarshal(response.Body.Bytes(), &decoded); err != nil {
t.Fatalf("decode response: %v", err)
}
if decoded.PaymentAutomated || decoded.AuthorizedCount != 1 || decoded.Tasks[0].AuthorizationID != result.Tasks[0].AuthorizationID {
t.Fatalf("unsafe or unstable response = %#v", decoded)
}
assertSecurityHeaders(t, response)
}
if store.startCalls != 2 || len(store.createdBy) != 2 || store.createdBy[0] != "admin" || store.createdBy[1] != "admin" {
t.Fatalf("store calls/created_by = %d/%#v", store.startCalls, store.createdBy)
}
for _, command := range store.commands {
if command.StartKey != startKeyForHTTP || len(command.Tasks) != 1 || command.Tasks[0].TaskID != taskIDForHTTP || command.Tasks[0].ExpectedTaskVersion != 7 {
t.Fatalf("decoded command = %#v", command)
}
}
}
func TestStartPurchasesMapsStoreErrorsWithoutLeakingDetails(t *testing.T) {
for name, test := range map[string]struct {
err error
want int
}{
"invalid": {err: tasks.ErrInvalidStart, want: http.StatusBadRequest},
"conflict": {err: tasks.ErrStartConflict, want: http.StatusConflict},
"internal": {err: errors.New("sqlite secret path and query"), want: http.StatusInternalServerError},
} {
t.Run(name, func(t *testing.T) {
store := &startRecordingStore{startErr: test.err}
router, _ := newRouterWithStore(t, store)
cookie, csrf := authenticatedStartSession(t, router)
response := serveStartRequest(t, router, validStartBody(), "application/json", csrf, cookie)
if response.Code != test.want || store.startCalls != 1 {
t.Fatalf("status/calls = %d/%d, want %d/1", response.Code, store.startCalls, test.want)
}
if response.Body.Len() != 0 || strings.Contains(response.Body.String(), "sqlite") {
t.Fatalf("error leaked details: %q", response.Body.String())
}
})
}
}
type startRecordingStore struct {
startResult tasks.StartResult
startErr error
startCalls int
commands []tasks.StartCommand
createdBy []string
}
func (store *startRecordingStore) CreateDraft(_ context.Context, draft tasks.Draft) (tasks.Draft, error) {
return draft, nil
}
func (store *startRecordingStore) ListDrafts(context.Context) ([]tasks.Draft, error) {
return nil, nil
}
func (store *startRecordingStore) ListTasks(context.Context, tasks.TaskFilter) ([]tasks.TaskRow, error) {
return nil, nil
}
func (store *startRecordingStore) StartPurchases(_ context.Context, command tasks.StartCommand, createdBy string) (tasks.StartResult, error) {
store.startCalls++
store.commands = append(store.commands, command)
store.createdBy = append(store.createdBy, createdBy)
return store.startResult, store.startErr
}
func authenticatedStartSession(t *testing.T, router http.Handler) (*http.Cookie, string) {
t.Helper()
cookie := authenticate(t, router)
page := serve(router, http.MethodGet, "/tasks", nil, cookie)
if page.Code != http.StatusOK {
t.Fatalf("GET /tasks status = %d", page.Code)
}
return cookie, csrfToken(t, page.Body.String())
}
func newStartRequest(t *testing.T, body, contentType, csrf string, cookie *http.Cookie) *http.Request {
t.Helper()
return newStartByteRequest(t, []byte(body), contentType, csrf, cookie)
}
func newStartByteRequest(t *testing.T, body []byte, contentType, csrf string, cookie *http.Cookie) *http.Request {
t.Helper()
request := httptest.NewRequest(http.MethodPost, "/tasks/start-purchases", bytes.NewReader(body))
if contentType != "" {
request.Header.Set("Content-Type", contentType)
}
if csrf != "" {
request.Header.Set("X-CSRF-Token", csrf)
}
if cookie != nil {
request.AddCookie(cookie)
}
return request
}
func serveStartBytes(t *testing.T, router http.Handler, body []byte, contentType, csrf string, cookie *http.Cookie) *httptest.ResponseRecorder {
t.Helper()
response := httptest.NewRecorder()
router.ServeHTTP(response, newStartByteRequest(t, body, contentType, csrf, cookie))
return response
}
func serveStartRequest(t *testing.T, router http.Handler, body, contentType, csrf string, cookie *http.Cookie) *httptest.ResponseRecorder {
t.Helper()
response := httptest.NewRecorder()
router.ServeHTTP(response, newStartRequest(t, body, contentType, csrf, cookie))
return response
}
func validStartBody() string {
return `{"start_key":"` + startKeyForHTTP + `","tasks":[{"task_id":"` + taskIDForHTTP + `","expected_task_version":7}]}`
}
func successfulStartResult() tasks.StartResult {
expires := time.Date(2026, 8, 4, 2, 3, 4, 0, time.UTC)
return tasks.StartResult{
StartKey: startKeyForHTTP,
AuthorizedCount: 1,
PaymentAutomated: false,
Tasks: []tasks.AuthorizedTask{{
TaskID: taskIDForHTTP,
TaskVersion: 8,
AuthorizationID: "d3c9f507-7473-4fa6-8d71-8786c34c6301",
ExpiresAt: expires,
}},
}
}
+128
View File
@@ -0,0 +1,128 @@
package tasks
import (
"errors"
"math"
"math/big"
"sort"
"strings"
"time"
_ "time/tzdata"
)
const maxStartItems = 100
var (
ErrStartConflict = errors.New("purchase start conflicts with current task state")
ErrInvalidStart = errors.New("invalid purchase start request")
)
type StartPolicy struct {
AuthorizationTTL time.Duration
MaxQuantity int
MaxTotalPrice string
}
type StartItem struct {
TaskID string `json:"task_id"`
ExpectedTaskVersion int `json:"expected_task_version"`
}
type StartCommand struct {
StartKey string `json:"start_key"`
Tasks []StartItem `json:"tasks"`
}
type AuthorizedTask struct {
TaskID string `json:"task_id"`
TaskVersion int `json:"task_version"`
AuthorizationID string `json:"authorization_id"`
ExpiresAt time.Time `json:"expires_at"`
}
type StartResult struct {
StartKey string `json:"start_key"`
AuthorizedCount int `json:"authorized_count"`
Tasks []AuthorizedTask `json:"tasks"`
PaymentAutomated bool `json:"payment_automated"`
}
type TaskFilter struct{ Keyword, Status, CreatedFrom, CreatedTo string }
type TaskRow struct {
ID, Title, GoodsID, SKUColor, SKUSize, MaxTotalPrice, Status string
Quantity, Version int
CreatedAt time.Time
}
func normalizeCents(value string) (string, *big.Int, bool) {
if value == "" || strings.TrimSpace(value) != value {
return "", nil, false
}
parts := strings.Split(value, ".")
if len(parts) != 2 || len(parts[0]) == 0 || len(parts[1]) != 2 || (len(parts[0]) > 1 && parts[0][0] == '0') {
return "", nil, false
}
for _, part := range parts {
for _, ch := range part {
if ch < '0' || ch > '9' {
return "", nil, false
}
}
}
cents := new(big.Int)
if _, ok := cents.SetString(parts[0]+parts[1], 10); !ok || cents.Sign() <= 0 {
return "", nil, false
}
return value, cents, true
}
func startItems(command StartCommand) ([]StartItem, error) {
if !validUUID(command.StartKey) || len(command.Tasks) == 0 || len(command.Tasks) > maxStartItems {
return nil, ErrInvalidStart
}
items := append([]StartItem(nil), command.Tasks...)
sort.Slice(items, func(i, j int) bool { return items[i].TaskID < items[j].TaskID })
for i, item := range items {
if !validUUID(item.TaskID) || item.ExpectedTaskVersion <= 0 || item.ExpectedTaskVersion == math.MaxInt || (i > 0 && item.TaskID == items[i-1].TaskID) {
return nil, ErrInvalidStart
}
}
return items, nil
}
func validTaskStatus(value string) bool {
if value == "" {
return true
}
for _, status := range []string{"DRAFT", "PENDING", "CLAIMED", "ORDERING", "NEEDS_MANUAL", "WAITING_PAYMENT", "RECONCILIATION_REQUIRED", "SUCCEEDED", "FAILED", "CANCELED"} {
if value == status {
return true
}
}
return false
}
func ShanghaiRange(from, to string) (time.Time, time.Time, error) {
if from == "" && to == "" {
return time.Time{}, time.Time{}, nil
}
location, err := time.LoadLocation("Asia/Shanghai")
if err != nil {
return time.Time{}, time.Time{}, err
}
parse := func(value string) (time.Time, error) { return time.ParseInLocation("2006-01-02", value, location) }
var start, end time.Time
if from != "" {
start, err = parse(from)
if err != nil {
return time.Time{}, time.Time{}, ErrInvalidStart
}
start = start.UTC()
}
if to != "" {
end, err = parse(to)
if err != nil {
return time.Time{}, time.Time{}, ErrInvalidStart
}
end = end.AddDate(0, 0, 1).UTC()
}
if !start.IsZero() && !end.IsZero() && !start.Before(end) {
return time.Time{}, time.Time{}, ErrInvalidStart
}
return start, end, nil
}
@@ -0,0 +1,424 @@
package tasks
import (
"context"
"database/sql"
"errors"
"fmt"
"math"
"reflect"
"sort"
"sync"
"testing"
"time"
"cmbuyer/admin/internal/migrations"
)
var fixedStartTime = time.Date(2026, 8, 4, 9, 2, 3, 456000000, time.FixedZone("UTC+8", 8*60*60))
func TestStartPurchasesPersistsCompleteSnapshotsForOneAndHundredTasks(t *testing.T) {
for _, count := range []int{1, 100} {
t.Run(fmt.Sprintf("%d tasks", count), func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
store.now = func() time.Time { return fixedStartTime }
items := make([]StartItem, 0, count)
wantDrafts := make(map[string]Draft, count)
for index := 1; index <= count; index++ {
id := startTestUUID(index)
draft := Draft{
ID: id,
Title: fmt.Sprintf("task-%03d", index),
GoodsID: fmt.Sprintf("937122%06d", index),
SKUColor: fmt.Sprintf("color-%03d", index),
SKUSize: fmt.Sprintf("size-%03d", index),
Quantity: index%10 + 1,
MaxTotalPrice: fmt.Sprintf("%d.%02d", index+10, index%100),
}
if _, err := store.CreateDraft(context.Background(), draft); err != nil {
t.Fatalf("create draft %d: %v", index, err)
}
items = append(items, StartItem{TaskID: id, ExpectedTaskVersion: 1})
wantDrafts[id] = draft
}
sort.Slice(items, func(i, j int) bool { return items[i].TaskID > items[j].TaskID })
command := StartCommand{StartKey: startTestUUID(1001 + count), Tasks: items}
result, err := store.StartPurchases(context.Background(), command, "authenticated-admin")
if err != nil {
t.Fatalf("StartPurchases: %v", err)
}
if result.StartKey != command.StartKey || result.AuthorizedCount != count || result.PaymentAutomated || len(result.Tasks) != count {
t.Fatalf("result = %#v", result)
}
wantCreated := fixedStartTime.UTC()
wantExpires := wantCreated.Add(15 * time.Minute)
seenAuthorizationIDs := map[string]bool{}
for index, authorized := range result.Tasks {
if index > 0 && result.Tasks[index-1].TaskID >= authorized.TaskID {
t.Fatalf("result is not in canonical task order: %#v", result.Tasks)
}
if authorized.TaskVersion != 2 || !authorized.ExpiresAt.Equal(wantExpires) || !validUUID(authorized.AuthorizationID) || seenAuthorizationIDs[authorized.AuthorizationID] {
t.Fatalf("authorized task = %#v", authorized)
}
seenAuthorizationIDs[authorized.AuthorizationID] = true
want := wantDrafts[authorized.TaskID]
var taskStatus, taskUpdated, authTaskID, authStartKey, goodsID, color, size, priceCap, authStatus, createdBy, createdAt, expiresAt string
var taskVersion, authTaskVersion, quantity int
err := database.QueryRow(`
SELECT t.status,t.version,t.updated_at,
a.task_id,a.task_version,a.start_key,a.goods_id,a.sku_color,a.sku_size,a.quantity,a.total_price_cap,a.status,a.created_by,a.created_at,a.expires_at
FROM tasks t JOIN order_authorizations a ON a.task_id=t.id WHERE a.id=?`, authorized.AuthorizationID).
Scan(&taskStatus, &taskVersion, &taskUpdated, &authTaskID, &authTaskVersion, &authStartKey, &goodsID, &color, &size, &quantity, &priceCap, &authStatus, &createdBy, &createdAt, &expiresAt)
if err != nil {
t.Fatalf("read authorization snapshot: %v", err)
}
if taskStatus != "PENDING" || taskVersion != 2 || taskUpdated != wantCreated.Format(time.RFC3339Nano) ||
authTaskID != want.ID || authTaskVersion != 2 || authStartKey != command.StartKey ||
goodsID != want.GoodsID || color != want.SKUColor || size != want.SKUSize || quantity != want.Quantity || priceCap != want.MaxTotalPrice ||
authStatus != "ACTIVE" || createdBy != "authenticated-admin" || createdAt != wantCreated.Format(time.RFC3339Nano) || expiresAt != wantExpires.Format(time.RFC3339Nano) {
t.Fatalf("stored task/authorization mismatch for %s", want.ID)
}
}
var distinctCreated, distinctExpires int
if err := database.QueryRow(`SELECT COUNT(DISTINCT created_at), COUNT(DISTINCT expires_at) FROM order_authorizations WHERE start_key=?`, command.StartKey).Scan(&distinctCreated, &distinctExpires); err != nil {
t.Fatal(err)
}
if distinctCreated != 1 || distinctExpires != 1 {
t.Fatalf("batch timestamps are not shared: created=%d expires=%d", distinctCreated, distinctExpires)
}
})
}
}
func TestStartPurchasesRejectsInvalidCommandsAndPolicyWithoutWrites(t *testing.T) {
validItem := StartItem{TaskID: startTestUUID(1), ExpectedTaskVersion: 1}
hundredOne := make([]StartItem, 101)
for index := range hundredOne {
hundredOne[index] = StartItem{TaskID: startTestUUID(index + 1), ExpectedTaskVersion: 1}
}
for name, command := range map[string]StartCommand{
"invalid start key": {StartKey: "not-a-uuid", Tasks: []StartItem{validItem}},
"empty tasks": {StartKey: startTestUUID(1001)},
"over batch limit": {StartKey: startTestUUID(1001), Tasks: hundredOne},
"invalid task id": {StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: "1", ExpectedTaskVersion: 1}}},
"duplicate task": {StartKey: startTestUUID(1001), Tasks: []StartItem{validItem, validItem}},
"zero version": {StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: validItem.TaskID}}},
"overflow version": {StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: validItem.TaskID, ExpectedTaskVersion: math.MaxInt}}},
} {
t.Run(name, func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
_, err := store.StartPurchases(context.Background(), command, "admin")
if !errors.Is(err, ErrInvalidStart) {
t.Fatalf("error = %v, want ErrInvalidStart", err)
}
assertAuthorizationCount(t, database, 0)
})
}
for name, mutate := range map[string]func(*SQLiteStore){
"zero ttl": func(store *SQLiteStore) { store.policy.AuthorizationTTL = 0 },
"zero quantity": func(store *SQLiteStore) { store.policy.MaxQuantity = 0 },
"bad max price": func(store *SQLiteStore) { store.policy.MaxTotalPrice = "999" },
} {
t.Run(name, func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
createStartDraft(t, store, validItem.TaskID)
mutate(store)
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{validItem}}, "admin")
if !errors.Is(err, ErrInvalidStart) {
t.Fatalf("error = %v, want ErrInvalidStart", err)
}
assertDraftUnchanged(t, database, validItem.TaskID)
assertAuthorizationCount(t, database, 0)
})
}
database := migratedDatabase(t)
store := configuredStartStore(t, database)
createStartDraft(t, store, validItem.TaskID)
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{validItem}}, "")
if !errors.Is(err, ErrInvalidStart) {
t.Fatalf("empty created_by error = %v", err)
}
assertDraftUnchanged(t, database, validItem.TaskID)
}
func TestStartPurchasesRejectsEveryTaskConflictWithoutAuthorization(t *testing.T) {
for name, mutate := range map[string]func(*testing.T, *SQLiteStore, string, *StartItem){
"missing": func(_ *testing.T, _ *SQLiteStore, _ string, item *StartItem) {
item.TaskID = startTestUUID(99)
},
"not draft": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET status='PENDING' WHERE id=?`, id)
},
"version mismatch": func(_ *testing.T, _ *SQLiteStore, _ string, item *StartItem) {
item.ExpectedTaskVersion = 2
},
"empty goods id": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET goods_id='' WHERE id=?`, id)
},
"nondigit goods id": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET goods_id='937x' WHERE id=?`, id)
},
"empty color": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET sku_color='' WHERE id=?`, id)
},
"empty size": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET sku_size='' WHERE id=?`, id)
},
"quantity over policy": func(_ *testing.T, store *SQLiteStore, _ string, _ *StartItem) {
store.policy.MaxQuantity = 1
},
"noncanonical price one decimal": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET max_total_price='12.8' WHERE id=?`, id)
},
"noncanonical leading zero": func(t *testing.T, store *SQLiteStore, id string, _ *StartItem) {
execTestSQL(t, store.database, `UPDATE tasks SET max_total_price='012.80' WHERE id=?`, id)
},
"price over policy": func(_ *testing.T, store *SQLiteStore, _ string, _ *StartItem) {
store.policy.MaxTotalPrice = "12.79"
},
} {
t.Run(name, func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
id := startTestUUID(1)
createStartDraft(t, store, id)
item := StartItem{TaskID: id, ExpectedTaskVersion: 1}
mutate(t, store, id, &item)
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{item}}, "admin")
if !errors.Is(err, ErrStartConflict) {
t.Fatalf("error = %v, want ErrStartConflict", err)
}
assertAuthorizationCount(t, database, 0)
})
}
}
func TestStartPurchasesRollsBackWholeBatchForLateConflictAndSQLFailure(t *testing.T) {
for _, test := range []struct {
name string
breakBatch func(*testing.T, *SQLiteStore, string)
}{
{name: "late validation conflict", breakBatch: func(t *testing.T, store *SQLiteStore, secondID string) {
execTestSQL(t, store.database, `UPDATE tasks SET sku_size='' WHERE id=?`, secondID)
}},
{name: "late SQL failure", breakBatch: func(t *testing.T, store *SQLiteStore, secondID string) {
statement := fmt.Sprintf(`CREATE TRIGGER reject_second_authorization BEFORE INSERT ON order_authorizations WHEN NEW.task_id='%s' BEGIN SELECT RAISE(ABORT, 'test failure'); END`, secondID)
execTestSQL(t, store.database, statement)
}},
} {
t.Run(test.name, func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
firstID, secondID := startTestUUID(1), startTestUUID(2)
createStartDraft(t, store, firstID)
createStartDraft(t, store, secondID)
test.breakBatch(t, store, secondID)
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: firstID, ExpectedTaskVersion: 1}, {TaskID: secondID, ExpectedTaskVersion: 1}}}, "admin")
if err == nil {
t.Fatal("StartPurchases unexpectedly succeeded")
}
assertDraftUnchanged(t, database, firstID)
var secondStatus string
var secondVersion int
if err := database.QueryRow(`SELECT status,version FROM tasks WHERE id=?`, secondID).Scan(&secondStatus, &secondVersion); err != nil {
t.Fatal(err)
}
if secondStatus != "DRAFT" || secondVersion != 1 {
t.Fatalf("second task = %s/v%d, want DRAFT/v1", secondStatus, secondVersion)
}
assertAuthorizationCount(t, database, 0)
})
}
}
func TestStartPurchasesReplayIsStableAndRejectsDifferentOrIncompleteSets(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
firstID, secondID, thirdID := startTestUUID(1), startTestUUID(2), startTestUUID(3)
for _, id := range []string{firstID, secondID, thirdID} {
createStartDraft(t, store, id)
}
command := StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: secondID, ExpectedTaskVersion: 1}, {TaskID: firstID, ExpectedTaskVersion: 1}}}
first, err := store.StartPurchases(context.Background(), command, "admin")
if err != nil {
t.Fatal(err)
}
command.Tasks[0], command.Tasks[1] = command.Tasks[1], command.Tasks[0]
replay, err := store.StartPurchases(context.Background(), command, "admin")
if err != nil || !reflect.DeepEqual(replay, first) {
t.Fatalf("replay = (%#v, %v), want %#v", replay, err, first)
}
assertAuthorizationCount(t, database, 2)
conflicting := []StartCommand{
{StartKey: command.StartKey, Tasks: command.Tasks[:1]},
{StartKey: command.StartKey, Tasks: []StartItem{{TaskID: firstID, ExpectedTaskVersion: 2}, {TaskID: secondID, ExpectedTaskVersion: 1}}},
{StartKey: command.StartKey, Tasks: []StartItem{{TaskID: firstID, ExpectedTaskVersion: 1}, {TaskID: secondID, ExpectedTaskVersion: 1}, {TaskID: thirdID, ExpectedTaskVersion: 1}}},
}
for _, changed := range conflicting {
if _, err := store.StartPurchases(context.Background(), changed, "admin"); !errors.Is(err, ErrStartConflict) {
t.Fatalf("different payload error = %v", err)
}
}
assertAuthorizationCount(t, database, 2)
assertDraftUnchanged(t, database, thirdID)
execTestSQL(t, database, `DELETE FROM order_authorizations WHERE task_id=?`, secondID)
if _, err := store.StartPurchases(context.Background(), command, "admin"); !errors.Is(err, ErrStartConflict) {
t.Fatalf("incomplete replay error = %v", err)
}
assertAuthorizationCount(t, database, 1)
}
func TestStartPurchasesConcurrentReplayAndVersionRace(t *testing.T) {
t.Run("same key replays one stable result", func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
id := startTestUUID(1)
createStartDraft(t, store, id)
command := StartCommand{StartKey: startTestUUID(1001), Tasks: []StartItem{{TaskID: id, ExpectedTaskVersion: 1}}}
const callers = 16
start := make(chan struct{})
results := make(chan StartResult, callers)
errorsChannel := make(chan error, callers)
var group sync.WaitGroup
for range callers {
group.Add(1)
go func() {
defer group.Done()
<-start
result, err := store.StartPurchases(context.Background(), command, "admin")
if err != nil {
errorsChannel <- err
return
}
results <- result
}()
}
close(start)
group.Wait()
close(results)
close(errorsChannel)
for err := range errorsChannel {
t.Fatalf("concurrent replay: %v", err)
}
var want StartResult
for result := range results {
if want.StartKey == "" {
want = result
} else if !reflect.DeepEqual(result, want) {
t.Fatalf("unstable replay: %#v != %#v", result, want)
}
}
assertAuthorizationCount(t, database, 1)
var version int
if err := database.QueryRow(`SELECT version FROM tasks WHERE id=?`, id).Scan(&version); err != nil || version != 2 {
t.Fatalf("task version = %d, err=%v", version, err)
}
})
t.Run("different keys race one expected version", func(t *testing.T) {
database := migratedDatabase(t)
store := configuredStartStore(t, database)
id := startTestUUID(1)
createStartDraft(t, store, id)
start := make(chan struct{})
errorsChannel := make(chan error, 2)
var group sync.WaitGroup
for _, key := range []string{startTestUUID(1001), startTestUUID(1002)} {
group.Add(1)
go func(startKey string) {
defer group.Done()
<-start
_, err := store.StartPurchases(context.Background(), StartCommand{StartKey: startKey, Tasks: []StartItem{{TaskID: id, ExpectedTaskVersion: 1}}}, "admin")
errorsChannel <- err
}(key)
}
close(start)
group.Wait()
close(errorsChannel)
successes, conflicts := 0, 0
for err := range errorsChannel {
switch {
case err == nil:
successes++
case errors.Is(err, ErrStartConflict):
conflicts++
default:
t.Fatalf("unexpected race error: %v", err)
}
}
if successes != 1 || conflicts != 1 {
t.Fatalf("success/conflict = %d/%d, want 1/1", successes, conflicts)
}
assertAuthorizationCount(t, database, 1)
})
}
func TestSQLiteStoreRejectsV1SchemaAtStartup(t *testing.T) {
database := openDatabase(t)
if err := migrations.Run(context.Background(), database, migrationDirectory(t), "up-by-one"); err != nil {
t.Fatalf("migrate to v1: %v", err)
}
if _, err := NewSQLiteStore(database); err == nil {
t.Fatal("NewSQLiteStore accepted the v1 two-pass schema")
}
}
func configuredStartStore(t *testing.T, database *sql.DB) *SQLiteStore {
t.Helper()
store, err := NewSQLiteStore(database)
if err != nil {
t.Fatalf("NewSQLiteStore: %v", err)
}
store.SetStartPolicy(StartPolicy{AuthorizationTTL: 15 * time.Minute, MaxQuantity: 10, MaxTotalPrice: "999.99"})
return store
}
func createStartDraft(t *testing.T, store *SQLiteStore, id string) {
t.Helper()
draft := Draft{ID: id, Title: "test", GoodsID: "937122477375", SKUColor: "黑色", SKUSize: "M", Quantity: 2, MaxTotalPrice: "12.80"}
if _, err := store.CreateDraft(context.Background(), draft); err != nil {
t.Fatalf("CreateDraft: %v", err)
}
}
func startTestUUID(number int) string {
return fmt.Sprintf("%08x-1234-4abc-a123-%012x", number, number)
}
func assertAuthorizationCount(t *testing.T, database *sql.DB, want int) {
t.Helper()
var got int
if err := database.QueryRow(`SELECT COUNT(*) FROM order_authorizations`).Scan(&got); err != nil {
t.Fatal(err)
}
if got != want {
t.Fatalf("authorization count = %d, want %d", got, want)
}
}
func assertDraftUnchanged(t *testing.T, database *sql.DB, id string) {
t.Helper()
var status string
var version int
if err := database.QueryRow(`SELECT status,version FROM tasks WHERE id=?`, id).Scan(&status, &version); err != nil {
t.Fatal(err)
}
if status != "DRAFT" || version != 1 {
t.Fatalf("task %s = %s/v%d, want DRAFT/v1", id, status, version)
}
}
func execTestSQL(t *testing.T, database *sql.DB, statement string, arguments ...any) {
t.Helper()
if _, err := database.Exec(statement, arguments...); err != nil {
t.Fatalf("execute test SQL: %v", err)
}
}
+229
View File
@@ -0,0 +1,229 @@
package tasks
import (
"context"
"database/sql"
"errors"
"strings"
"time"
"cmbuyer/admin/internal/domain"
)
// ErrInvalidFilter 表示任务筛选值无效,路由应按字段重新渲染而不是泄露内部错误。
var ErrInvalidFilter = errors.New("invalid task filter")
// SetStartPolicy is called during startup; policy is explicit because authorization limits must not be implicit defaults.
func (store *SQLiteStore) SetStartPolicy(policy StartPolicy) { store.policy = policy }
func (store *SQLiteStore) ListTasks(ctx context.Context, filter TaskFilter) ([]TaskRow, error) {
if !ValidateTaskFilter(filter).Valid() {
return nil, ErrInvalidFilter
}
from, to, err := ShanghaiRange(filter.CreatedFrom, filter.CreatedTo)
if err != nil {
return nil, ErrInvalidFilter
}
clauses, args := []string{"1=1"}, []any{}
if filter.Status != "" {
clauses = append(clauses, "status = ?")
args = append(args, filter.Status)
}
if filter.Keyword != "" {
escaped := strings.NewReplacer("\\", "\\\\", "%", "\\%", "_", "\\_").Replace(filter.Keyword)
clauses = append(clauses, "(title LIKE ? ESCAPE '\\' OR goods_id LIKE ? ESCAPE '\\')")
args = append(args, "%"+escaped+"%", "%"+escaped+"%")
}
if !from.IsZero() {
clauses = append(clauses, "julianday(created_at) >= julianday(?)")
args = append(args, from.Format(time.RFC3339Nano))
}
if !to.IsZero() {
clauses = append(clauses, "julianday(created_at) < julianday(?)")
args = append(args, to.Format(time.RFC3339Nano))
}
rows, err := store.database.QueryContext(ctx, "SELECT id,title,goods_id,sku_color,sku_size,quantity,max_total_price,status,version,created_at FROM tasks WHERE "+strings.Join(clauses, " AND ")+" ORDER BY julianday(created_at) DESC,rowid DESC", args...)
if err != nil {
return nil, err
}
defer rows.Close()
result := []TaskRow{}
for rows.Next() {
var item TaskRow
var created string
if err := rows.Scan(&item.ID, &item.Title, &item.GoodsID, &item.SKUColor, &item.SKUSize, &item.Quantity, &item.MaxTotalPrice, &item.Status, &item.Version, &created); err != nil {
return nil, err
}
item.CreatedAt, err = time.Parse(time.RFC3339Nano, created)
if err != nil {
return nil, err
}
result = append(result, item)
}
return result, rows.Err()
}
// ValidateTaskFilter 返回可关联到字段的错误,使服务端页面拒绝篡改参数时仍能保留输入值。
func ValidateTaskFilter(filter TaskFilter) Errors {
validation := Errors{}
if !validTaskStatus(filter.Status) {
validation["status"] = "请选择有效的任务状态。"
}
location, err := time.LoadLocation("Asia/Shanghai")
if err != nil {
validation["created_from"] = "日期筛选暂不可用,请稍后重试。"
validation["created_to"] = "日期筛选暂不可用,请稍后重试。"
return validation
}
parseDate := func(field, value string) (time.Time, bool) {
if value == "" {
return time.Time{}, true
}
parsed, parseErr := time.ParseInLocation("2006-01-02", value, location)
if parseErr != nil {
validation[field] = "请输入有效日期。"
return time.Time{}, false
}
return parsed, true
}
from, fromOK := parseDate("created_from", filter.CreatedFrom)
to, toOK := parseDate("created_to", filter.CreatedTo)
if fromOK && toOK && !from.IsZero() && !to.IsZero() && from.After(to) {
validation["created_to"] = "结束日期不能早于开始日期。"
}
return validation
}
func (store *SQLiteStore) StartPurchases(ctx context.Context, command StartCommand, createdBy string) (StartResult, error) {
items, err := startItems(command)
if err != nil || createdBy == "" {
return StartResult{}, ErrInvalidStart
}
if store.policy.AuthorizationTTL <= 0 || store.policy.MaxQuantity <= 0 {
return StartResult{}, ErrInvalidStart
}
_, ceiling, ok := normalizeCents(store.policy.MaxTotalPrice)
if !ok {
return StartResult{}, ErrInvalidStart
}
writeCtx, cancel := context.WithTimeout(ctx, sqliteWriteTimeout)
defer cancel()
select {
case store.writeGate <- struct{}{}:
defer func() { <-store.writeGate }()
case <-writeCtx.Done():
return StartResult{}, writeCtx.Err()
}
tx, err := store.database.BeginTx(writeCtx, nil)
if err != nil {
return StartResult{}, err
}
defer tx.Rollback()
// Replay precedes any DRAFT check. One service process serializes this check with creation; SQLite uniqueness remains the cross-transaction backstop.
result, found, err := replayStart(writeCtx, tx, command.StartKey, items)
if err != nil {
return StartResult{}, err
}
if found {
if err := tx.Commit(); err != nil {
return StartResult{}, err
}
return result, nil
}
now := store.now().UTC()
expires := now.Add(store.policy.AuthorizationTTL)
result = StartResult{StartKey: command.StartKey, AuthorizedCount: len(items), Tasks: make([]AuthorizedTask, 0, len(items)), PaymentAutomated: false}
for _, item := range items {
var title, goods, color, size, price, status string
var quantity, version int
if err := tx.QueryRowContext(writeCtx, "SELECT title,goods_id,sku_color,sku_size,quantity,max_total_price,status,version FROM tasks WHERE id=?", item.TaskID).Scan(&title, &goods, &color, &size, &quantity, &price, &status, &version); err != nil {
if err == sql.ErrNoRows {
return StartResult{}, ErrStartConflict
}
return StartResult{}, err
}
if status != "DRAFT" || version != item.ExpectedTaskVersion || !goodsIDValid(goods) || color == "" || size == "" || quantity < 1 || quantity > store.policy.MaxQuantity {
return StartResult{}, ErrStartConflict
}
canonical, cents, ok := normalizeCents(price)
if !ok || canonical != price || cents.Cmp(ceiling) > 0 {
return StartResult{}, ErrStartConflict
}
if _, err := domain.TransitionTask(domain.TaskStatusDraft, domain.TaskStatusPending); err != nil {
return StartResult{}, err
}
id, err := NewCreateKey()
if err != nil {
return StartResult{}, err
}
next := version + 1
if _, err = tx.ExecContext(writeCtx, "INSERT INTO order_authorizations (id,task_id,task_version,start_key,goods_id,sku_color,sku_size,quantity,total_price_cap,status,created_by,created_at,expires_at) VALUES (?,?,?,?,?,?,?,?,?,'ACTIVE',?,?,?)", id, item.TaskID, next, command.StartKey, goods, color, size, quantity, price, createdBy, now.Format(time.RFC3339Nano), expires.Format(time.RFC3339Nano)); err != nil {
return StartResult{}, err
}
updated, err := tx.ExecContext(writeCtx, "UPDATE tasks SET status='PENDING',version=version+1,updated_at=? WHERE id=? AND status='DRAFT' AND version=?", now.Format(time.RFC3339Nano), item.TaskID, version)
if err != nil {
return StartResult{}, err
}
affected, err := updated.RowsAffected()
if err != nil {
return StartResult{}, err
}
if affected != 1 {
return StartResult{}, ErrStartConflict
}
result.Tasks = append(result.Tasks, AuthorizedTask{TaskID: item.TaskID, TaskVersion: next, AuthorizationID: id, ExpiresAt: expires})
}
if err := tx.Commit(); err != nil {
return StartResult{}, err
}
return result, nil
}
func goodsIDValid(value string) bool {
if value == "" {
return false
}
for _, ch := range value {
if ch < '0' || ch > '9' {
return false
}
}
return true
}
func replayStart(ctx context.Context, tx *sql.Tx, startKey string, items []StartItem) (StartResult, bool, error) {
rows, err := tx.QueryContext(ctx, "SELECT id,task_id,task_version,expires_at FROM order_authorizations WHERE start_key=? ORDER BY task_id", startKey)
if err != nil {
return StartResult{}, false, err
}
defer rows.Close()
result := StartResult{StartKey: startKey, PaymentAutomated: false}
for rows.Next() {
var item AuthorizedTask
var expires string
if err := rows.Scan(&item.AuthorizationID, &item.TaskID, &item.TaskVersion, &expires); err != nil {
return StartResult{}, false, err
}
item.ExpiresAt, err = time.Parse(time.RFC3339Nano, expires)
if err != nil {
return StartResult{}, false, err
}
result.Tasks = append(result.Tasks, item)
}
if err := rows.Err(); err != nil {
return StartResult{}, false, err
}
if len(result.Tasks) == 0 {
return StartResult{}, false, nil
}
if len(result.Tasks) != len(items) {
return StartResult{}, false, ErrStartConflict
}
for i := range items {
if result.Tasks[i].TaskID != items[i].TaskID || result.Tasks[i].TaskVersion-1 != items[i].ExpectedTaskVersion {
return StartResult{}, false, ErrStartConflict
}
}
result.AuthorizedCount = len(result.Tasks)
return result, true, nil
}
+195
View File
@@ -0,0 +1,195 @@
package tasks
import (
"context"
"database/sql"
"errors"
"testing"
"time"
)
func TestListTasksTreatsLikeMetacharactersLiterally(t *testing.T) {
database := migratedDatabase(t)
store, err := NewSQLiteStore(database)
if err != nil {
t.Fatal(err)
}
created := "2026-08-04T01:00:00Z"
insertTaskRow(t, database, "percent", "100%纯棉", "100", "DRAFT", created)
insertTaskRow(t, database, "underscore", "尺码_A", "101", "DRAFT", created)
insertTaskRow(t, database, "backslash", `路径\名称`, "102", "DRAFT", created)
insertTaskRow(t, database, "plain", "普通商品", "103", "DRAFT", created)
for _, test := range []struct {
keyword string
wantID string
}{
{keyword: "%", wantID: "percent"},
{keyword: "_", wantID: "underscore"},
{keyword: `\`, wantID: "backslash"},
} {
t.Run(test.wantID, func(t *testing.T) {
rows, err := store.ListTasks(context.Background(), TaskFilter{Keyword: test.keyword})
if err != nil {
t.Fatal(err)
}
if len(rows) != 1 || rows[0].ID != test.wantID {
t.Fatalf("keyword %q rows = %#v, want only %q", test.keyword, rows, test.wantID)
}
})
}
}
func TestListTasksSupportsEveryStatusAndEmptyMeansAll(t *testing.T) {
database := migratedDatabase(t)
store, err := NewSQLiteStore(database)
if err != nil {
t.Fatal(err)
}
statuses := []string{"DRAFT", "PENDING", "CLAIMED", "ORDERING", "NEEDS_MANUAL", "WAITING_PAYMENT", "RECONCILIATION_REQUIRED", "SUCCEEDED", "FAILED", "CANCELED"}
for index, status := range statuses {
insertTaskRow(t, database, status, status, "200", status, time.Date(2026, 8, 4, 1, 0, index, 0, time.UTC).Format(time.RFC3339Nano))
}
all, err := store.ListTasks(context.Background(), TaskFilter{})
if err != nil {
t.Fatal(err)
}
if len(all) != len(statuses) {
t.Fatalf("all-status rows = %d, want %d", len(all), len(statuses))
}
for _, status := range statuses {
rows, err := store.ListTasks(context.Background(), TaskFilter{Status: status})
if err != nil {
t.Fatalf("status %s: %v", status, err)
}
if len(rows) != 1 || rows[0].Status != status {
t.Fatalf("status %s rows = %#v", status, rows)
}
}
}
func TestListTasksUsesShanghaiHalfOpenDateRange(t *testing.T) {
database := migratedDatabase(t)
store, err := NewSQLiteStore(database)
if err != nil {
t.Fatal(err)
}
insertTaskRow(t, database, "before", "before", "300", "DRAFT", "2026-08-03T15:59:59Z")
insertTaskRow(t, database, "at-start", "at-start", "301", "DRAFT", "2026-08-03T16:00:00Z")
insertTaskRow(t, database, "before-end", "before-end", "302", "DRAFT", "2026-08-04T15:59:59Z")
insertTaskRow(t, database, "at-end", "at-end", "303", "DRAFT", "2026-08-04T16:00:00Z")
rows, err := store.ListTasks(context.Background(), TaskFilter{CreatedFrom: "2026-08-04", CreatedTo: "2026-08-04"})
if err != nil {
t.Fatal(err)
}
if len(rows) != 2 || rows[0].ID != "before-end" || rows[1].ID != "at-start" {
t.Fatalf("Shanghai day rows = %#v, want [before-end at-start]", rows)
}
}
func TestListTasksBreaksEqualTimestampsByDescendingRowID(t *testing.T) {
database := migratedDatabase(t)
store, err := NewSQLiteStore(database)
if err != nil {
t.Fatal(err)
}
created := "2026-08-04T01:02:03Z"
insertTaskRow(t, database, "first", "first", "400", "DRAFT", created)
insertTaskRow(t, database, "second", "second", "401", "DRAFT", created)
rows, err := store.ListTasks(context.Background(), TaskFilter{})
if err != nil {
t.Fatal(err)
}
if len(rows) != 2 || rows[0].ID != "second" || rows[1].ID != "first" {
t.Fatalf("equal-time rows = %#v, want descending rowid", rows)
}
}
func TestListTasksRejectsInvalidStatusAndDates(t *testing.T) {
store, err := NewSQLiteStore(migratedDatabase(t))
if err != nil {
t.Fatal(err)
}
for name, filter := range map[string]TaskFilter{
"status": {Status: "UNKNOWN"},
"from date": {CreatedFrom: "2026-02-30"},
"to date": {CreatedTo: "04/08/2026"},
"reverse range": {CreatedFrom: "2026-08-05", CreatedTo: "2026-08-04"},
} {
t.Run(name, func(t *testing.T) {
rows, err := store.ListTasks(context.Background(), filter)
if !errors.Is(err, ErrInvalidFilter) || rows != nil {
t.Fatalf("ListTasks(%#v) = (%#v, %v), want ErrInvalidFilter", filter, rows, err)
}
})
}
}
func TestStartPurchasesIsAtomicAndReplaysSameSet(t *testing.T) {
database := migratedDatabase(t)
store, err := NewSQLiteStore(database)
if err != nil {
t.Fatal(err)
}
store.SetStartPolicy(StartPolicy{AuthorizationTTL: time.Hour, MaxQuantity: 10, MaxTotalPrice: "999.99"})
store.now = func() time.Time { return time.Date(2026, 8, 4, 1, 2, 3, 0, time.UTC) }
for _, draft := range []Draft{testDraft(testKey, "one"), testDraft("b3c9f507-7473-4fa6-8d71-8786c34c6301", "two")} {
if _, err := store.CreateDraft(context.Background(), draft); err != nil {
t.Fatal(err)
}
}
command := StartCommand{StartKey: "c3c9f507-7473-4fa6-8d71-8786c34c6301", Tasks: []StartItem{{TaskID: "b3c9f507-7473-4fa6-8d71-8786c34c6301", ExpectedTaskVersion: 1}, {TaskID: testKey, ExpectedTaskVersion: 1}}}
first, err := store.StartPurchases(context.Background(), command, "admin")
if err != nil {
t.Fatal(err)
}
if first.AuthorizedCount != 2 || first.PaymentAutomated {
t.Fatalf("start result=%#v", first)
}
command.Tasks[0], command.Tasks[1] = command.Tasks[1], command.Tasks[0]
replay, err := store.StartPurchases(context.Background(), command, "admin")
if err != nil {
t.Fatal(err)
}
if replay.Tasks[0].AuthorizationID != first.Tasks[0].AuthorizationID || replay.Tasks[1].AuthorizationID != first.Tasks[1].AuthorizationID {
t.Fatalf("replay=%#v first=%#v", replay, first)
}
var pending, auths int
if err := database.QueryRow(`SELECT COUNT(*) FROM tasks WHERE status='PENDING' AND version=2`).Scan(&pending); err != nil || pending != 2 {
t.Fatalf("pending=%d err=%v", pending, err)
}
if err := database.QueryRow(`SELECT COUNT(*) FROM order_authorizations WHERE status='ACTIVE' AND created_by='admin'`).Scan(&auths); err != nil || auths != 2 {
t.Fatalf("auths=%d err=%v", auths, err)
}
_, err = store.StartPurchases(context.Background(), StartCommand{StartKey: command.StartKey, Tasks: command.Tasks[:1]}, "admin")
if !errors.Is(err, ErrStartConflict) {
t.Fatalf("subset err=%v", err)
}
}
func TestShanghaiRangeAndMoneyAreFailClosed(t *testing.T) {
start, end, err := ShanghaiRange("2026-08-04", "2026-08-04")
if err != nil || start.Format(time.RFC3339) != "2026-08-03T16:00:00Z" || end.Format(time.RFC3339) != "2026-08-04T16:00:00Z" {
t.Fatalf("range=(%s,%s,%v)", start, end, err)
}
for _, value := range []string{"0.01", "12.80", "999999999999999999999999.99"} {
if _, _, ok := normalizeCents(value); !ok {
t.Fatalf("money %q rejected", value)
}
}
for _, value := range []string{"1", "01.20", "0.00", "1.234", "1.", " 1.00", "1e2"} {
if _, _, ok := normalizeCents(value); ok {
t.Fatalf("money %q accepted", value)
}
}
}
func insertTaskRow(t *testing.T, database *sql.DB, id, title, goodsID, status, createdAt string) {
t.Helper()
if _, err := database.Exec(`INSERT INTO tasks (id, source, title, goods_id, sku_color, sku_size, quantity, max_total_price, status, version, created_at, updated_at) VALUES (?, 'MANUAL', ?, ?, '黑色', 'M', 2, '12.80', ?, 1, ?, ?)`, id, title, goodsID, status, createdAt, createdAt); err != nil {
t.Fatalf("insert task %s: %v", id, err)
}
}
+13 -7
View File
@@ -13,21 +13,27 @@ const sqliteWriteTimeout = 2 * time.Second
type Store interface {
CreateDraft(context.Context, Draft) (Draft, error)
ListDrafts(context.Context) ([]Draft, error)
ListTasks(context.Context, TaskFilter) ([]TaskRow, error)
StartPurchases(context.Context, StartCommand, string) (StartResult, error)
}
type SQLiteStore struct {
database *sql.DB
now func() time.Time
createGate chan struct{}
database *sql.DB
now func() time.Time
writeGate chan struct{}
policy StartPolicy
}
func NewSQLiteStore(database *sql.DB) (*SQLiteStore, error) {
if database == nil {
return nil, errors.New("database is required")
}
if _, err := database.Exec("SELECT 1 FROM tasks LIMIT 1"); err != nil {
if _, err := database.Exec("SELECT task_version, start_key, total_price_cap FROM order_authorizations LIMIT 1"); err != nil {
return nil, fmt.Errorf("tasks migration is not available: %w", err)
}
return &SQLiteStore{database: database, now: time.Now, createGate: make(chan struct{}, 1)}, nil
if _, err := database.Exec("SELECT 1 FROM purchase_attempts LIMIT 1"); err != nil {
return nil, fmt.Errorf("single-pass migration is not available: %w", err)
}
return &SQLiteStore{database: database, now: time.Now, writeGate: make(chan struct{}, 1)}, nil
}
func (store *SQLiteStore) CreateDraft(ctx context.Context, draft Draft) (Draft, error) {
@@ -36,8 +42,8 @@ func (store *SQLiteStore) CreateDraft(ctx context.Context, draft Draft) (Draft,
// SQLite permits one writer at a time. Serializing this store's short create
// transaction prevents concurrent retries of one create key from surfacing as busy.
select {
case store.createGate <- struct{}{}:
defer func() { <-store.createGate }()
case store.writeGate <- struct{}{}:
defer func() { <-store.writeGate }()
case <-writeContext.Done():
return Draft{}, writeContext.Err()
}
@@ -0,0 +1,60 @@
(() => {
"use strict";
const form = document.querySelector("[data-start-purchases]");
if (!form) return;
const all = form.querySelector("[data-select-all]");
const summary = form.querySelector("[data-selection-summary]");
const button = form.querySelector("[data-start-button]");
const feedback = form.querySelector("[data-start-feedback]");
const boxes = () => [...form.querySelectorAll("input[data-task-id]")];
let selectionFrozen = false;
const parseCents = (value) => {
const match = /^(0|[1-9]\d*)\.(\d{2})$/.exec(value);
return match ? BigInt(match[1] + match[2]) : null;
};
const refresh = () => {
const available = boxes();
const selected = available.filter((box) => box.checked);
let cents = 0n;
let pricesValid = true;
selected.forEach((box) => {
const price = parseCents(box.dataset.price);
if (price === null) pricesValid = false;
else cents += price;
});
summary.textContent = `已选 ${selected.length} 条,最高总额 ¥${cents / 100n}.${(cents % 100n).toString().padStart(2, "0")}`;
button.disabled = !selected.length || !pricesValid;
if (!pricesValid) feedback.textContent = "所选任务金额无法安全汇总,请刷新后重选。";
if (all) {
all.checked = selected.length > 0 && selected.length === available.length;
all.indeterminate = selected.length > 0 && selected.length < available.length;
all.disabled = selectionFrozen || available.length === 0;
}
};
const freezeSelection = (frozen) => {
selectionFrozen = frozen;
boxes().forEach((box) => { box.disabled = frozen; });
refresh();
};
boxes().forEach((box) => box.addEventListener("change", refresh));
if (all) all.addEventListener("change", () => { boxes().forEach((box) => { box.checked = all.checked; }); refresh(); });
let frozenPayload = null;
let inFlight = false;
form.addEventListener("submit", async (event) => {
event.preventDefault();
const selected = boxes().filter((box) => box.checked);
if (!selected.length || inFlight) return;
const tasks = selected.map((box) => ({task_id: box.dataset.taskId, expected_task_version: Number(box.dataset.taskVersion)}));
if (tasks.some((item) => !Number.isSafeInteger(item.expected_task_version) || item.expected_task_version < 1)) { feedback.textContent = "任务版本无效,请刷新后重选。"; return; }
frozenPayload = frozenPayload || JSON.stringify({start_key: form.dataset.startKey, tasks});
inFlight = true; freezeSelection(true); button.disabled = true; button.textContent = "正在授权…";
try { const response = await fetch("/tasks/start-purchases", {method:"POST", headers:{"Content-Type":"application/json", "X-CSRF-Token":form.dataset.csrf}, body:frozenPayload});
if (response.ok) { window.location.reload(); return; }
if (response.status === 409) { feedback.textContent = "任务已变化,请刷新后重选。"; frozenPayload = null; freezeSelection(false); boxes().forEach((box) => { box.checked = false; }); refresh(); return; }
if (response.status === 400 || response.status === 401 || response.status === 403) { feedback.textContent = "请求未被接受,请刷新页面后重试。"; frozenPayload = null; freezeSelection(false); return; }
feedback.textContent = "结果暂时不明确,只能使用同一按钮原样重放。";
} catch (_) { feedback.textContent = "网络结果不明确,请使用同一按钮原样重试。"; }
finally { inFlight = false; button.textContent = "开始采购(只创建待付款订单)"; if (frozenPayload) button.disabled = false; }
});
refresh();
})();
@@ -0,0 +1,138 @@
"use strict";
const test = require("node:test");
const assert = require("node:assert/strict");
const fs = require("node:fs");
const path = require("node:path");
const vm = require("node:vm");
const source = fs.readFileSync(path.join(__dirname, "tasks.js"), "utf8");
test("successful authorization sends numeric version and reloads", async () => {
const requests = [];
const harness = createHarness(async (_url, options) => {
requests.push(options);
return {ok: true, status: 200};
});
await harness.submit();
assert.equal(requests.length, 1);
assert.equal(requests[0].headers["Content-Type"], "application/json");
assert.equal(requests[0].headers["X-CSRF-Token"], "csrf-token");
const payload = JSON.parse(requests[0].body);
assert.equal(payload.start_key, "start-key");
assert.equal(typeof payload.tasks[0].expected_task_version, "number");
assert.equal(payload.tasks[0].expected_task_version, 7);
assert.equal(harness.reloads(), 1);
});
test("409 clears stale selection and requires a fresh choice", async () => {
const harness = createHarness(async () => ({ok: false, status: 409}));
await harness.submit();
assert.equal(harness.box.checked, false);
assert.equal(harness.box.disabled, false);
assert.equal(harness.button.disabled, true);
assert.match(harness.feedback.textContent, /任务已变化/);
});
for (const status of [400, 401, 403]) {
test(`${status} releases the frozen payload for a page refresh`, async () => {
const harness = createHarness(async () => ({ok: false, status}));
await harness.submit();
assert.equal(harness.box.checked, true);
assert.equal(harness.box.disabled, false);
assert.equal(harness.button.disabled, false);
assert.match(harness.feedback.textContent, /刷新页面后重试/);
});
}
test("5xx retries the byte-identical frozen payload", async () => {
const bodies = [];
const harness = createHarness(async (_url, options) => {
bodies.push(options.body);
return {ok: false, status: 503};
});
await harness.submit();
assert.equal(harness.box.disabled, true);
assert.equal(harness.button.disabled, false);
assert.match(harness.feedback.textContent, /原样重放/);
await harness.submit();
assert.equal(bodies.length, 2);
assert.equal(bodies[1], bodies[0]);
});
test("network ambiguity retries the same payload and can finish", async () => {
const bodies = [];
let call = 0;
const harness = createHarness(async (_url, options) => {
bodies.push(options.body);
call++;
if (call === 1) throw new Error("network result unknown");
return {ok: true, status: 200};
});
await harness.submit();
assert.equal(harness.box.disabled, true);
assert.match(harness.feedback.textContent, /原样重试/);
await harness.submit();
assert.deepEqual(bodies, [bodies[0], bodies[0]]);
assert.equal(harness.reloads(), 1);
});
function createHarness(fetchImplementation) {
class FakeElement {
constructor() {
this.dataset = {};
this.checked = false;
this.disabled = false;
this.indeterminate = false;
this.textContent = "";
this.listeners = {};
}
addEventListener(type, listener) {
this.listeners[type] = listener;
}
}
const box = new FakeElement();
box.checked = true;
box.dataset = {taskId: "task-id", taskVersion: "7", price: "12.80"};
const selectAll = new FakeElement();
const summary = new FakeElement();
const button = new FakeElement();
const feedback = new FakeElement();
const form = new FakeElement();
form.dataset = {startKey: "start-key", csrf: "csrf-token"};
form.querySelector = (selector) => ({
"[data-select-all]": selectAll,
"[data-selection-summary]": summary,
"[data-start-button]": button,
"[data-start-feedback]": feedback,
})[selector] || null;
form.querySelectorAll = (selector) => selector === "input[data-task-id]" ? [box] : [];
let reloadCount = 0;
const context = {
document: {querySelector: (selector) => selector === "[data-start-purchases]" ? form : null},
fetch: fetchImplementation,
window: {location: {reload: () => { reloadCount++; }}},
};
vm.runInNewContext(source, context, {filename: "tasks.js"});
return {
box,
button,
feedback,
reloads: () => reloadCount,
submit: () => form.listeners.submit({preventDefault() {}}),
};
}
@@ -6,12 +6,27 @@
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>采购任务 · 采购服务</title>
<style>
:root{--bg:#f4f7fb;--surface:#fff;--text:#172033;--muted:#526079;--border:#cfd8e6;--primary:#155eef;--danger:#b42318;--success:#067647;--focus:#ffbf47;font-family:"Segoe UI","Microsoft YaHei UI",system-ui,sans-serif}*{box-sizing:border-box}html{min-width:320px;background:var(--bg)}body{min-height:100dvh;margin:0;color:var(--text);background:var(--bg);font-size:16px;line-height:1.55}button,input{font:inherit}:focus-visible{outline:3px solid var(--focus);outline-offset:3px}.skip{position:fixed;z-index:100;top:8px;left:8px;padding:10px;color:#fff;background:#172033;transform:translateY(-160%)}.skip:focus{transform:translateY(0)}header{display:flex;align-items:center;justify-content:space-between;gap:16px;min-height:64px;padding:10px clamp(16px,4vw,40px);border-bottom:1px solid var(--border);background:var(--surface)}.brand{font-weight:700}.brand b{display:inline-grid;place-items:center;width:32px;height:32px;margin-right:8px;border-radius:8px;background:var(--primary);color:#fff;font-size:.82rem}.logout,.button{display:inline-flex;align-items:center;justify-content:center;min-height:44px;padding:9px 14px;border:1px solid var(--border);border-radius:8px;color:var(--text);background:#fff;font-weight:700;text-decoration:none;cursor:pointer}.button.primary{border-color:var(--primary);background:var(--primary);color:#fff}.button:disabled,.filter input:disabled{opacity:.5;cursor:not-allowed}main{width:min(100% - 32px,1200px);margin:32px auto}.toolbar{display:flex;align-items:center;justify-content:space-between;gap:16px;margin-bottom:16px}.toolbar-actions,.filters,.actions{display:flex;flex-wrap:wrap;gap:10px}.muted,.placeholder{color:var(--muted)}.filters{align-items:end;margin:0 0 16px}.filters label{display:grid;gap:4px;font-weight:700}.filters input{min-height:44px;min-width:180px;padding:8px 10px;border:1px solid var(--border);border-radius:8px;background:#fff}.table-wrap{overflow-x:auto;border:1px solid var(--border);border-radius:12px;background:var(--surface)}table{width:100%;min-width:880px;border-collapse:collapse}th,td{padding:12px 14px;border-bottom:1px solid var(--border);text-align:left;vertical-align:top}th{background:#f8fafc;font-size:.88rem}td a{color:#124cc5;font-weight:700;text-underline-offset:3px}.status{display:inline-block;padding:3px 8px;border-radius:999px;background:#eaf1ff;color:#173d8f;font-size:.85rem;font-weight:700}.empty,.success{padding:20px;border:1px solid var(--border);border-radius:12px;background:var(--surface)}.success{margin:0 0 16px;border-color:#9dd9b8;background:#ecfdf3;color:var(--success)}.modal-scrim{position:fixed;z-index:20;inset:0;background:rgba(23,32,51,.52)}dialog[open]{position:fixed;z-index:30;top:50%;left:50%;width:min(calc(100% - 24px),640px);max-height:calc(100dvh - 24px);margin:0;padding:28px;overflow-y:auto;border:1px solid var(--border);border-radius:14px;box-shadow:0 18px 48px rgba(23,32,51,.24);transform:translate(-50%,-50%);background:var(--surface)}.form-page{width:min(100% - 32px,640px);margin:32px auto;padding:28px;border:1px solid var(--border);border-radius:14px;background:var(--surface)}.form-grid{display:grid;gap:16px}.field label{display:block;margin-bottom:6px;font-weight:700}.required{color:var(--danger)}.field input{width:100%;min-height:44px;padding:10px 12px;border:1px solid #9ba9bc;border-radius:8px}.field input[aria-invalid=true]{border-color:var(--danger)}.error{margin:5px 0 0;color:var(--danger);font-size:.9rem}.summary{margin:0 0 16px;padding:12px;border-left:4px solid var(--danger);background:#fef3f2;color:var(--danger)}.summary p{margin:0}.summary ul{margin:8px 0 0;padding-left:20px}.summary a{color:inherit}.sr-only{position:absolute;width:1px;height:1px;padding:0;margin:-1px;overflow:hidden;clip:rect(0,0,0,0);white-space:nowrap;border:0}@media(max-width:420px){main,.form-page{width:calc(100% - 24px);margin:24px auto}.toolbar{align-items:stretch;flex-direction:column}.toolbar-actions,.toolbar .button{width:100%}.toolbar-actions .button{flex:1}.filters{align-items:stretch;flex-direction:column}.filters input,.filters .button{width:100%}}@media(prefers-reduced-motion:reduce){*,*::before,*::after{transition-duration:.01ms!important;animation-duration:.01ms!important}}</style>
:root{--bg:#f4f7fb;--surface:#fff;--text:#172033;--muted:#526079;--border:#cfd8e6;--primary:#155eef;--danger:#b42318;--success:#067647;--focus:#ffbf47;font-family:"Segoe UI","Microsoft YaHei UI",system-ui,sans-serif}*{box-sizing:border-box}html{min-width:320px;background:var(--bg)}body{min-height:100dvh;margin:0;color:var(--text);background:var(--bg);font-size:16px;line-height:1.55}button,input,select{font:inherit}:focus-visible{outline:3px solid var(--focus);outline-offset:3px}.skip{position:fixed;z-index:100;top:8px;left:8px;padding:10px;color:#fff;background:#172033;transform:translateY(-160%)}.skip:focus{transform:translateY(0)}header{display:flex;align-items:center;justify-content:space-between;gap:16px;min-height:64px;padding:10px clamp(16px,4vw,40px);border-bottom:1px solid var(--border);background:var(--surface)}.brand{font-weight:700}.brand b{display:inline-grid;place-items:center;width:32px;height:32px;margin-right:8px;border-radius:8px;background:var(--primary);color:#fff;font-size:.82rem}.logout,.button{display:inline-flex;align-items:center;justify-content:center;min-height:44px;padding:9px 14px;border:1px solid var(--border);border-radius:8px;color:var(--text);background:#fff;font-weight:700;text-decoration:none;cursor:pointer}.button.primary{border-color:var(--primary);background:var(--primary);color:#fff}.button:disabled,.filters input:disabled,.filters select:disabled{opacity:.5;cursor:not-allowed}main{width:min(100% - 32px,1200px);margin:32px auto}.toolbar{display:flex;align-items:center;justify-content:space-between;gap:16px;margin-bottom:16px}.toolbar-actions,.filters,.actions,.batch-actions{display:flex;flex-wrap:wrap;gap:10px}.muted,.placeholder,.not-selectable{color:var(--muted)}.filters{align-items:end;margin:0 0 16px}.filter-field{display:grid;gap:4px}.filter-field label{font-weight:700}.filters input,.filters select{min-height:44px;min-width:180px;padding:8px 10px;border:1px solid var(--border);border-radius:8px;background:#fff}.filters [aria-invalid=true]{border-color:var(--danger)}.batch-bar{display:flex;align-items:center;justify-content:space-between;gap:16px;margin:0 0 16px;padding:14px 16px;border:1px solid var(--border);border-radius:12px;background:var(--surface)}.batch-bar p{margin:2px 0}.batch-summary{font-weight:700}.batch-message{min-height:1.55em;color:var(--muted)}.table-wrap{overflow-x:auto;border:1px solid var(--border);border-radius:12px;background:var(--surface)}table{width:100%;min-width:880px;border-collapse:collapse}th,td{padding:12px 14px;border-bottom:1px solid var(--border);text-align:left;vertical-align:top}th{background:#f8fafc;font-size:.88rem}td a{color:#124cc5;font-weight:700;text-underline-offset:3px}.select-cell{width:64px;text-align:center}.checkbox-target{display:inline-grid;place-items:center;min-width:44px;min-height:44px;margin:-10px;cursor:pointer}.checkbox-target input{width:18px;height:18px}.status{display:inline-block;padding:3px 8px;border-radius:999px;background:#eaf1ff;color:#173d8f;font-size:.85rem;font-weight:700}.empty,.success{padding:20px;border:1px solid var(--border);border-radius:12px;background:var(--surface)}.success{margin:0 0 16px;border-color:#9dd9b8;background:#ecfdf3;color:var(--success)}.modal-scrim{position:fixed;z-index:20;inset:0;background:rgba(23,32,51,.52)}dialog[open]{position:fixed;z-index:30;top:50%;left:50%;width:min(calc(100% - 24px),640px);max-height:calc(100dvh - 24px);margin:0;padding:28px;overflow-y:auto;border:1px solid var(--border);border-radius:14px;box-shadow:0 18px 48px rgba(23,32,51,.24);transform:translate(-50%,-50%);background:var(--surface)}.form-page{width:min(100% - 32px,640px);margin:32px auto;padding:28px;border:1px solid var(--border);border-radius:14px;background:var(--surface)}.form-grid{display:grid;gap:16px}.field label{display:block;margin-bottom:6px;font-weight:700}.required{color:var(--danger)}.field input{width:100%;min-height:44px;padding:10px 12px;border:1px solid #9ba9bc;border-radius:8px}.field input[aria-invalid=true]{border-color:var(--danger)}.error{margin:5px 0 0;color:var(--danger);font-size:.9rem}.summary{margin:0 0 16px;padding:12px;border-left:4px solid var(--danger);background:#fef3f2;color:var(--danger)}.summary p{margin:0}.summary ul{margin:8px 0 0;padding-left:20px}.summary a{color:inherit}.sr-only{position:absolute;width:1px;height:1px;padding:0;margin:-1px;overflow:hidden;clip:rect(0,0,0,0);white-space:nowrap;border:0}@media(max-width:420px){main,.form-page{width:calc(100% - 24px);margin:24px auto}.toolbar,.batch-bar{align-items:stretch;flex-direction:column}.toolbar-actions,.toolbar .button,.batch-actions,.batch-actions .button{width:100%}.toolbar-actions .button,.batch-actions .button{flex:1}.filters{align-items:stretch;flex-direction:column}.filters input,.filters select,.filters .button{width:100%}}@media(prefers-reduced-motion:reduce){*,*::before,*::after{transition-duration:.01ms!important;animation-duration:.01ms!important}}</style>
</head>
<body>
<a class="skip" href="#main">跳到主要内容</a>
<header><div class="brand"><b aria-hidden="true">采</b>采购服务</div><form method="post" action="/logout"><input type="hidden" name="csrf_token" value="{{.CSRFToken}}"><button class="logout" type="submit">退出登录</button></form></header>
{{if .FullPage}}<main class="form-page" id="main">{{template "form" .}}</main>{{else}}<main id="main"><div class="toolbar"><div><h1>采购任务</h1><p class="muted">只显示待开始的手工任务。</p></div><div class="toolbar-actions"><button class="button" type="button" disabled>导入</button><a class="button primary" href="/tasks?create=1">创建任务</a></div></div><div class="filters" aria-label="暂不可用的列表条件"><label>关键词<input type="search" disabled></label><button class="button" type="button" disabled>筛选</button><button class="button" type="button" disabled>清除</button></div>{{if .Success}}<p class="success" role="status">任务已创建,已显示在列表首行。</p>{{end}}{{if .Drafts}}<div class="table-wrap"><table><thead><tr><th scope="col"><input type="checkbox" disabled aria-label="选择全部任务"></th><th scope="col">标题</th><th scope="col">颜色分类</th><th scope="col">尺码</th><th scope="col">价格上限</th><th scope="col">数量</th><th scope="col">采购结果</th><th scope="col">状态</th><th scope="col">创建时间</th></tr></thead><tbody>{{range .Drafts}}<tr><td><input type="checkbox" disabled aria-label="选择任务 {{.Title}}"></td><td><a href="https://mobile.yangkeduo.com/goods.html?goods_id={{.GoodsID}}" target="_blank" rel="noopener noreferrer">{{.Title}}</a></td><td>{{.SKUColor}}</td><td>{{.SKUSize}}</td><td>¥{{.MaxTotalPrice}}</td><td>{{.Quantity}}</td><td>—</td><td><span class="status">待开始</span></td><td><time datetime="{{.CreatedAt.Format "2006-01-02T15:04:05Z07:00"}}">{{.CreatedAt.Format "2006-01-02 15:04 UTC"}}</time></td></tr>{{end}}</tbody></table></div>{{else}}<section class="empty"><h2>还没有待开始任务</h2><p>创建一条手工任务后会显示在这里。</p></section>{{end}}</main>{{if .OpenForm}}<div class="modal-scrim" aria-hidden="true"></div><dialog open aria-modal="true" aria-labelledby="form-title">{{template "form" .}}</dialog>{{end}}{{end}}
{{if .FullPage}}<main class="form-page" id="main">{{template "form" .}}</main>{{else}}<main id="main">
<div class="toolbar"><div><h1>采购任务</h1><p class="muted">查询任务并统一授权待开始任务。</p></div><div class="toolbar-actions"><button class="button" type="button" disabled>导入</button><a class="button primary" href="/tasks?create=1">创建任务</a></div></div>
{{if .FilterErrors}}<div class="summary" role="alert" aria-live="assertive"><p>请修正筛选条件后重新查询。</p><ul>{{with index .FilterErrors "status"}}<li><a href="#filter-status">状态:{{.}}</a></li>{{end}}{{with index .FilterErrors "created_from"}}<li><a href="#filter-created-from">开始日期:{{.}}</a></li>{{end}}{{with index .FilterErrors "created_to"}}<li><a href="#filter-created-to">结束日期:{{.}}</a></li>{{end}}</ul></div>{{end}}
<form class="filters" method="get" action="/tasks" aria-label="任务筛选">
<div class="filter-field"><label for="filter-keyword">关键词</label><input id="filter-keyword" name="keyword" type="search" value="{{.Filter.Keyword}}" placeholder="标题或商品编号"></div>
<div class="filter-field"><label for="filter-status">状态</label><select id="filter-status" name="status" aria-invalid="{{if index .FilterErrors "status"}}true{{else}}false{{end}}"{{with index .FilterErrors "status"}} aria-describedby="filter-status-error"{{end}}>{{if index .FilterErrors "status"}}<option value="{{.Filter.Status}}" selected>无效状态:{{.Filter.Status}}</option>{{end}}<option value=""{{if eq .Filter.Status ""}} selected{{end}}>全部状态</option><option value="DRAFT"{{if eq .Filter.Status "DRAFT"}} selected{{end}}>待开始</option><option value="PENDING"{{if eq .Filter.Status "PENDING"}} selected{{end}}>已授权待领取</option><option value="CLAIMED"{{if eq .Filter.Status "CLAIMED"}} selected{{end}}>已领取</option><option value="ORDERING"{{if eq .Filter.Status "ORDERING"}} selected{{end}}>执行中</option><option value="NEEDS_MANUAL"{{if eq .Filter.Status "NEEDS_MANUAL"}} selected{{end}}>待人工处理</option><option value="WAITING_PAYMENT"{{if eq .Filter.Status "WAITING_PAYMENT"}} selected{{end}}>待付款</option><option value="RECONCILIATION_REQUIRED"{{if eq .Filter.Status "RECONCILIATION_REQUIRED"}} selected{{end}}>围栏后待调和</option><option value="SUCCEEDED"{{if eq .Filter.Status "SUCCEEDED"}} selected{{end}}>已完成</option><option value="FAILED"{{if eq .Filter.Status "FAILED"}} selected{{end}}>失败</option><option value="CANCELED"{{if eq .Filter.Status "CANCELED"}} selected{{end}}>已取消</option></select>{{with index .FilterErrors "status"}}<p class="error" id="filter-status-error">{{.}}</p>{{end}}</div>
<div class="filter-field"><label for="filter-created-from">开始日期</label><input id="filter-created-from" name="created_from" type="date" value="{{.Filter.CreatedFrom}}" aria-invalid="{{if index .FilterErrors "created_from"}}true{{else}}false{{end}}"{{with index .FilterErrors "created_from"}} aria-describedby="filter-created-from-error"{{end}}>{{with index .FilterErrors "created_from"}}<p class="error" id="filter-created-from-error">{{.}}</p>{{end}}</div>
<div class="filter-field"><label for="filter-created-to">结束日期</label><input id="filter-created-to" name="created_to" type="date" value="{{.Filter.CreatedTo}}" aria-invalid="{{if index .FilterErrors "created_to"}}true{{else}}false{{end}}"{{with index .FilterErrors "created_to"}} aria-describedby="filter-created-to-error"{{end}}>{{with index .FilterErrors "created_to"}}<p class="error" id="filter-created-to-error">{{.}}</p>{{end}}</div>
<div class="actions"><button class="button primary" type="submit">筛选</button><a class="button" href="/tasks">清除筛选</a></div>
</form>
{{if .Success}}<p class="success" role="status">任务已创建,已显示在列表首行。</p>{{end}}
<form data-start-purchases data-start-key="{{.StartKey}}" data-csrf="{{.CSRFToken}}">
<section class="batch-bar" aria-label="批量开始采购"><div><p class="batch-summary" data-selection-summary aria-live="polite">已选 0 条,最高总额 ¥0.00</p><p class="muted" id="payment-note">采购工具会逐条创建待付款订单,系统不会付款。</p><p class="batch-message" id="start-feedback" data-start-feedback role="status" aria-live="polite"></p></div><div class="batch-actions"><button class="button primary" type="submit" data-start-button aria-describedby="payment-note start-feedback" disabled>开始采购(只创建待付款订单)</button></div></section>
<div class="table-wrap"><table><thead><tr><th class="select-cell" scope="col"><label class="checkbox-target"><span class="sr-only">选择全部当前筛选结果中的待开始任务</span><input type="checkbox" data-select-all aria-label="选择全部任务"{{if not .Tasks}} disabled{{end}}></label></th><th scope="col">标题</th><th scope="col">颜色分类</th><th scope="col">尺码</th><th scope="col">价格上限</th><th scope="col">数量</th><th scope="col">采购结果</th><th scope="col">状态</th><th scope="col">创建时间(上海)</th></tr></thead><tbody>{{if .Tasks}}{{range .Tasks}}<tr><td class="select-cell">{{if eq .Status "DRAFT"}}<label class="checkbox-target"><span class="sr-only">选择任务 {{.Title}}</span><input type="checkbox" name="task_ids" value="{{.ID}}" data-task-id="{{.ID}}" data-task-version="{{.Version}}" data-price="{{.MaxTotalPrice}}" aria-label="选择任务 {{.Title}}"></label>{{else}}<span class="not-selectable">—<span class="sr-only">{{statusLabel .Status}}任务不可选择</span></span>{{end}}</td><td><a href="https://mobile.yangkeduo.com/goods.html?goods_id={{.GoodsID}}" target="_blank" rel="noopener noreferrer">{{.Title}}</a></td><td>{{.SKUColor}}</td><td>{{.SKUSize}}</td><td>¥{{.MaxTotalPrice}}</td><td>{{.Quantity}}</td><td>—</td><td><span class="status">{{statusLabel .Status}}</span></td><td><time datetime="{{shanghaiDateTime .CreatedAt}}">{{shanghaiTime .CreatedAt}}</time></td></tr>{{end}}{{else}}<tr><td colspan="9">{{if .FilterErrors}}<section class="empty"><h2>筛选条件有误</h2><p>请修正上方标出的字段后重新查询。</p></section>{{else if .HasFilter}}<section class="empty"><h2>没有符合筛选条件的任务</h2><p><a class="button" href="/tasks">清除筛选</a></p></section>{{else}}<section class="empty"><h2>还没有采购任务</h2><p>创建一条手工任务后会显示在这里。</p></section>{{end}}</td></tr>{{end}}</tbody></table></div>
</form>
</main>{{if .OpenForm}}<div class="modal-scrim" aria-hidden="true"></div><dialog open aria-modal="true" aria-labelledby="form-title">{{template "form" .}}</dialog>{{end}}<script src="/static/tasks.js" defer></script>{{end}}
</body>
</html>
{{end}}
+46 -10
View File
@@ -5,6 +5,7 @@ import (
"embed"
"html/template"
"io"
"time"
"cmbuyer/admin/internal/tasks"
)
@@ -12,7 +13,17 @@ import (
//go:embed templates/*.html
var templateFiles embed.FS
var templates = template.Must(template.New("webui").Funcs(template.FuncMap{"list": func(values ...any) []any { return values }}).ParseFS(templateFiles, "templates/*.html"))
//go:embed static/tasks.js
var tasksScript []byte
var shanghaiLocation = time.FixedZone("Asia/Shanghai", 8*60*60)
var templates = template.Must(template.New("webui").Funcs(template.FuncMap{
"list": func(values ...any) []any { return values },
"statusLabel": statusLabel,
"shanghaiDateTime": func(value time.Time) string { return value.In(shanghaiLocation).Format(time.RFC3339) },
"shanghaiTime": func(value time.Time) string { return value.In(shanghaiLocation).Format("2006-01-02 15:04") },
}).ParseFS(templateFiles, "templates/*.html"))
// LoginData 是登录页面所需的非敏感展示数据。
type LoginData struct {
@@ -22,16 +33,20 @@ type LoginData struct {
Error string
}
// TasksData 是受保护的 DRAFT 建单与列表页面所需数据。
// TasksData 是受保护的建单与任务工作台页面所需数据。
type TasksData struct {
CSRFToken string
Drafts []tasks.Draft
Form tasks.Form
Errors tasks.Errors
OpenForm bool
FullPage bool
FocusField string
Success bool
CSRFToken string
Tasks []tasks.TaskRow
Filter tasks.TaskFilter
FilterErrors tasks.Errors
HasFilter bool
StartKey string
Form tasks.Form
Errors tasks.Errors
OpenForm bool
FullPage bool
FocusField string
Success bool
}
// RenderLogin 写入登录页。
@@ -43,3 +58,24 @@ func RenderLogin(writer io.Writer, data LoginData) error {
func RenderTasks(writer io.Writer, data TasksData) error {
return templates.ExecuteTemplate(writer, "tasks.html", data)
}
func TasksScript() []byte { return tasksScript }
func statusLabel(status string) string {
labels := map[string]string{
"DRAFT": "待开始",
"PENDING": "已授权待领取",
"CLAIMED": "已领取",
"ORDERING": "执行中",
"NEEDS_MANUAL": "待人工处理",
"WAITING_PAYMENT": "待付款",
"RECONCILIATION_REQUIRED": "围栏后待调和",
"SUCCEEDED": "已完成",
"FAILED": "失败",
"CANCELED": "已取消",
}
if label, ok := labels[status]; ok {
return label
}
return "未知状态"
}
-84
View File
@@ -1,84 +0,0 @@
"""恢复 T-103 已取证目标规格、验证现价并保存本地原始截图。"""
from __future__ import annotations
import argparse
from math import isfinite
from pathlib import Path
import sys
CLIENT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(CLIENT_ROOT / "src"))
from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner
from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector
from cmbuyer_client.pdd.product_url import ProductUrlError, parse_product_url
from cmbuyer_client.pdd.sku_selection import EXPECTED_GOODS_ID, SkuSelectionError, TASK_TO_UI_SELECTION
from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError, SkuSelectionRunner
def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="恢复 T-103 已取证规格并保存本地原始截图。")
parser.add_argument("--serial", required=True, help="ADB device serial;禁止自动选择。")
parser.add_argument("--url", required=True, help="唯一 canonical goods.html?goods_id= 直链。")
parser.add_argument("--color", required=True, help="T-103 任务颜色值。")
parser.add_argument("--size", required=True, help="T-103 任务尺码值。")
parser.add_argument("--output-dir", required=True, type=Path, help="新建本地目录;不得覆盖已有目录。")
parser.add_argument("--timeout", type=float, default=10.0, help="ADB 与设备 RPC 超时(秒)。")
parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。")
return parser.parse_args(argv)
def validate_arguments(arguments: argparse.Namespace) -> None:
if not isinstance(arguments.serial, str) or not arguments.serial.strip():
raise ValueError("必须显式提供非空 --serial。")
if not isinstance(arguments.timeout, (int, float)) or isinstance(arguments.timeout, bool) or arguments.timeout <= 0 or not isfinite(arguments.timeout):
raise ValueError("--timeout 必须是大于 0 的有限数值。")
link = parse_product_url(arguments.url)
if link.goods_id != EXPECTED_GOODS_ID:
raise ValueError("--url 不是 T-103 已取证商品。")
if (arguments.color, arguments.size) not in TASK_TO_UI_SELECTION:
raise ValueError("--color 与 --size 必须是 T-103 已取证任务值。")
def main(argv: list[str] | None = None) -> int:
arguments = parse_arguments(argv)
try:
validate_arguments(arguments)
except (ValueError, ProductUrlError) as error:
print(f"失败:{error}", file=sys.stderr)
return 2
try:
import adbutils
import uiautomator2 as u2
except ImportError:
print("失败:缺少 uiautomator2;请在采购工具虚拟环境中运行。", file=sys.stderr)
return 2
runner = SkuSelectionRunner(
AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout),
NoReconnectUiautomatorConnector(adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, u2.connect),
timeout_seconds=arguments.timeout,
)
try:
result = runner.run(arguments.serial, arguments.url, arguments.color, arguments.size, arguments.output_dir)
except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError) as error:
# Flow 可能来自测试替身或未来实现;CLI 不回显任何异常正文,避免泄露节点树或页面文本。
print("规格恢复失败:已停止,未发布本地证据目录。", file=sys.stderr)
return 1
except OSError:
print("规格恢复失败:无法创建或发布本地证据目录。", file=sys.stderr)
return 1
print(f"规格恢复完成:{result.output_directory}")
print(f"manifest:{result.manifest_path}")
print(f"目标规格:{arguments.color} / {arguments.size}")
print(f"确认单价:{result.unit_price}")
print("页面对应性:请人工核对本地原始截图。")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+2 -10
View File
@@ -1,23 +1,15 @@
"""拼多多链接的受限打开、只读取证与经取证的规格面板选择。
"""拼多多链接的受限打开与只读取证。
此包不提供通用页面选择器、输入、滑动或任何订单动作。
此包不提供页面选择器、输入、滑动、下单或支付能力。
"""
from .product_open import ProductOpenCapturer, ProductOpenResult
from .product_url import ProductUrl, ProductUrlError, parse_product_url
from .sku_selection import SkuSelection, SkuSelectionError, SkuSelectionFlow
from .sku_selection_runner import SkuSelectionRunError, SkuSelectionRunResult, SkuSelectionRunner
__all__ = [
"ProductOpenCapturer",
"ProductOpenResult",
"ProductUrl",
"ProductUrlError",
"SkuSelection",
"SkuSelectionError",
"SkuSelectionFlow",
"SkuSelectionRunError",
"SkuSelectionRunResult",
"SkuSelectionRunner",
"parse_product_url",
]
@@ -1,420 +0,0 @@
"""T-103:仅限已取证 PDD 8.17.0 的规格面板恢复。"""
from __future__ import annotations
from dataclasses import dataclass
import re
from time import monotonic, sleep
from typing import Any, Callable, Protocol
from xml.etree import ElementTree
from ..device.baseline import PDD_PACKAGE
from .product_open import EXPECTED_PDD_VERSION
from .product_url import parse_product_url
EXPECTED_GOODS_ID = "937122477375"
EXPECTED_UNIT_PRICE = "12.88"
# 任务值不是页面判据;右侧是 v5 取证的唯一 accessibility 文案(空格/全角括号均有意义)。
TASK_TO_UI_SELECTION = {("黑色CHA(纯棉)", "M(建议100-115)"): ("黑色 CHA (纯棉)", "M(建议100-115)")}
_TARGET_COLOR_UI, _TARGET_SIZE_UI = next(iter(TASK_TO_UI_SELECTION.values()))
_ENTRY = "快要抢光"
_ENTRY_TEXT_BOUNDS = "[900,1312][1056,1355]"
_ENTRY_INNER_BOUNDS = "[712,1312][1056,1355]"
_ENTRY_ACTION_BOUNDS = "[0,1256][1080,1355]"
_SIZE = "尺码"
_W, _H = 1080, 2376
_PRICE_PARENT = "[396,498][895,570]"
_CURRENT = "[396,503][712,570]"
_ORIGINAL = "[730,503][895,570]"
_SUMMARY = "[396,654][1053,716]"
_COLOR_REGION = "[36,1000][1080,1631]"
_SIZE_LABEL = "[36,1654][114,1700]"
_SIZE_HEADER = "[36,1637][1044,1718]"
_SIZE_OPTIONS = "[36,1730][1044,2045]"
_BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
_PRICE = re.compile(r"^[^0-9¥¥]*[¥¥]([1-9][0-9]*\.[0-9]{2})$")
_ORIGINAL_PRICE = re.compile(r"^[¥¥][1-9][0-9]*\.[0-9]{2}$")
_BAD_PRICE_ROLE = ("提交订单", "支付", "优惠", "券", "会员", "补贴", "区间", "实付", "到手", "原价", "划线价", "最低", "低至", "起价", "下单", "先用后付", "预估")
class SkuSelectionError(RuntimeError):
"""已取证判据不成立时的脱敏停止。"""
class SkuPanelDevice(Protocol):
def app_info(self, package_name: str) -> dict[str, Any]: ...
def app_current(self) -> dict[str, Any]: ...
def dump_window_hierarchy(self) -> str: ...
def tap_sku_entry(self, bounds: str) -> None: ...
def tap_sku_option(self, bounds: str) -> None: ...
def leave_sku_panel(self) -> None: ...
@dataclass(frozen=True)
class SkuSelection:
color: str
size: str
def resolve_task_selection(color: str, size: str) -> SkuSelection:
mapped = TASK_TO_UI_SELECTION.get((color, size))
if mapped is None:
raise SkuSelectionError("规格任务值不是已取证的唯一目标,已停止操作。")
return SkuSelection(*mapped)
@dataclass(frozen=True)
class _Node:
element: ElementTree.Element
parent: "_Node | None"
@property
def text(self) -> str: return self.element.get("text", "")
@property
def desc(self) -> str: return self.element.get("content-desc", "")
@property
def bounds(self) -> str: return self.element.get("bounds", "")
class SkuSelectionFlow:
def __init__(self, device: SkuPanelDevice, entry_wait_timeout_seconds: float = 0.2,
entry_poll_interval_seconds: float = 0.2, monotonic_clock: Callable[[], float] = monotonic,
sleep_function: Callable[[float], None] = sleep) -> None:
if entry_wait_timeout_seconds < 0 or entry_poll_interval_seconds <= 0:
raise ValueError("入口等待参数无效。")
self._device, self._entry_timeout, self._poll = device, entry_wait_timeout_seconds, entry_poll_interval_seconds
self._clock, self._sleep = monotonic_clock, sleep_function
self._pending: tuple[str, Callable[[list[_Node]], Any]] | None = None
def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
if parse_product_url(product_url).goods_id != EXPECTED_GOODS_ID:
raise SkuSelectionError("商品不是已取证目标,已停止操作。")
if pre_intent_hierarchy is not None:
previous_nodes = _parse_nodes(pre_intent_hierarchy)
if _eligible_entries(previous_nodes):
raise SkuSelectionError("intent 前页面已出现规格入口,已拒绝旧商品误点。")
entry, before = self._wait_for_entry(pre_intent_hierarchy)
_action_bounds(entry.bounds)
self._pending = (before, _panel)
self._device.tap_sku_entry(entry.bounds)
self._wait_after_action(before, _panel)
def select_sku_options(self, selection: SkuSelection) -> None:
if selection not in {SkuSelection(*item) for item in TASK_TO_UI_SELECTION.values()}:
raise SkuSelectionError("规格 UI 文案不是获准目标,已停止操作。")
initial = self._verified_nodes()
_option(initial, "color", selection.color); _option(initial, "size", selection.size)
_selected_label(initial, "color"); _selected_label(initial, "size")
self._restore("color", selection.color)
self._restore("size", selection.size)
def read_sku_unit_price(self) -> str:
return _unit_price(self._verified_nodes())
def verify_target_selection_and_read_price(self, selection: SkuSelection) -> str:
nodes = self._verified_nodes()
_selected(nodes, "color", selection.color)
_selected(nodes, "size", selection.size)
return _unit_price(nodes)
def exit_sku_panel_safely(self) -> None:
self._require_foreground()
before = self._read_hierarchy()
_panel(_parse_nodes(before))
self._device.leave_sku_panel()
deadline = self._clock() + self._entry_timeout
while True:
self._require_foreground()
raw = self._read_hierarchy()
if raw != before:
try:
_panel(_parse_nodes(raw))
except SkuSelectionError:
return
remaining = deadline - self._clock()
if remaining <= 0:
raise SkuSelectionError("安全退出后未确认离开规格面板,未重试返回。")
self._sleep(min(self._poll, remaining))
def reconcile_pending_action(self) -> None:
"""仅只读调和一次已发出但尚未得到后置条件确认的动作。"""
if self._pending is None:
return
before, condition = self._pending
self._wait_after_action(before, condition)
def _restore(self, dimension: str, expected: str) -> None:
self._require_foreground()
before = self._read_hierarchy()
nodes = _panel(_parse_nodes(before))
target = _option(nodes, dimension, expected)
if _selected_label(nodes, dimension) == expected:
return
_action_bounds(target.bounds)
condition: Callable[[list[_Node]], Any]
if dimension == "color":
condition = lambda refreshed: _post_color(refreshed, expected)
else:
condition = lambda refreshed: _post_all_targets(refreshed, expected)
self._pending = (before, condition)
self._device.tap_sku_option(target.bounds)
if dimension == "color":
self._wait_after_action(before, condition)
else:
self._wait_after_action(before, condition)
def _wait_for_entry(self, previous: str | None) -> tuple[_Node, str]:
deadline, stable = self._clock() + self._entry_timeout, None
while True:
self._require_version()
current = self._device.app_current()
if isinstance(current, dict) and current.get("package") == PDD_PACKAGE:
raw = self._read_hierarchy()
entries = _eligible_entries(_parse_nodes(raw))
if len(entries) > 1:
raise SkuSelectionError("商品页规格入口不唯一,已停止操作。")
if len(entries) == 1 and raw != previous:
if stable == raw:
return entries[0], raw
stable = raw
else:
stable = None
else:
stable = None
remaining = deadline - self._clock()
if remaining <= 0:
raise SkuSelectionError("等待已取证规格入口超时,未执行点击。")
self._sleep(min(self._poll, remaining))
def _wait_after_action(self, previous: str, condition: Callable[[list[_Node]], Any]) -> list[_Node]:
deadline = self._clock() + self._entry_timeout
while True:
self._require_foreground()
raw = self._read_hierarchy()
if raw != previous:
nodes = _parse_nodes(raw)
try:
condition(nodes)
self._pending = None
return nodes
except SkuSelectionError:
pass
remaining = deadline - self._clock()
if remaining <= 0:
raise SkuSelectionError("动作后页面未在限定时间内满足已取证后置条件,未重试动作。")
self._sleep(min(self._poll, remaining))
def _verified_nodes(self) -> list[_Node]:
self._require_foreground()
return _panel(self._read_nodes())
def _require_version(self) -> None:
info = self._device.app_info(PDD_PACKAGE)
version = (info.get("versionName") or info.get("version_name")) if isinstance(info, dict) else None
if version != EXPECTED_PDD_VERSION:
raise SkuSelectionError("拼多多版本与已取证版本不一致,已停止操作。")
def _require_foreground(self) -> None:
self._require_version()
current = self._device.app_current()
if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
raise SkuSelectionError("拼多多不在前台,已停止操作。")
def _read_hierarchy(self) -> str:
try: raw = self._device.dump_window_hierarchy()
except Exception as error: raise SkuSelectionError("节点树读取失败,已停止操作。") from error
if not isinstance(raw, str) or not raw: raise SkuSelectionError("节点树不可用,已停止操作。")
return raw
def _read_nodes(self) -> list[_Node]: return _parse_nodes(self._read_hierarchy())
def _parse_nodes(raw: str) -> list[_Node]:
try: root = ElementTree.fromstring(raw)
except ElementTree.ParseError as error: raise SkuSelectionError("节点树格式无效,已停止操作。") from error
if root.tag != "hierarchy": raise SkuSelectionError("节点树根节点无效,已停止操作。")
result: list[_Node] = []
def visit(element: ElementTree.Element, parent: _Node | None) -> None:
node = _Node(element, parent); result.append(node)
for child in element: visit(child, node)
visit(root, None)
return result
def _panel(nodes: list[_Node]) -> list[_Node]:
parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一,已停止操作。")
_one([n for n in nodes if n.parent is parent and n.bounds == _ORIGINAL and _readonly(n) and _ORIGINAL_PRICE.fullmatch(n.text)], "规格面板原价槽位不唯一,已停止操作。")
_one([n for n in nodes if n.bounds == _SUMMARY and _readonly(n) and n.text.startswith("已选:")], "规格面板已选摘要不唯一,已停止操作。")
_color_container(nodes); _size_container(nodes)
return nodes
def _unit_price(nodes: list[_Node]) -> str:
parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一,已停止读取。")
money = [n for n in nodes if n.parent is parent and _readonly(n) and any(mark in n.text for mark in "¥¥")]
if len(money) != 2: raise SkuSelectionError("规格面板金额槽位不唯一,已停止读取。")
current = _one([n for n in money if n.bounds == _CURRENT and not _clickable_ancestor(n) and not any(word in n.text for word in _BAD_PRICE_ROLE) and _PRICE.fullmatch(n.text)], "规格面板现价不唯一或不符合已取证槽位,已停止读取。")
if not any(n.bounds == _ORIGINAL and _ORIGINAL_PRICE.fullmatch(n.text) for n in money):
raise SkuSelectionError("规格面板原价槽位无效,已停止读取。")
match = _PRICE.fullmatch(current.text)
if match is None: raise SkuSelectionError("规格面板现价格式失效,已停止读取。")
return match.group(1)
def _option(nodes: list[_Node], dimension: str, expected: str) -> _Node:
_panel(nodes)
return _one([n for n in _options(nodes, dimension) if _label(n) == expected], "规格选项不唯一或不是精确匹配,已停止操作。")
def _selected(nodes: list[_Node], dimension: str, expected: str) -> None:
if _selected_label(nodes, dimension) != expected:
raise SkuSelectionError("规格选择后读回的 selected 文案不一致,已停止操作。")
def _selected_label(nodes: list[_Node], dimension: str) -> str:
selected = [n for n in _options(nodes, dimension) if n.element.get("selected") == "true"]
label = _label(_one(selected, "规格维度没有唯一 selected 状态,已停止操作。"))
if label is None: raise SkuSelectionError("规格维度 selected 文案无效,已停止操作。")
return label
def _post_color(nodes: list[_Node], expected: str) -> None:
_panel(nodes)
_selected(nodes, "color", expected)
_selected_label(nodes, "size")
def _post_all_targets(nodes: list[_Node], expected_size: str) -> None:
_panel(nodes)
_selected(nodes, "color", _TARGET_COLOR_UI)
_selected(nodes, "size", expected_size)
def _options(nodes: list[_Node], dimension: str) -> list[_Node]:
container = _color_container(nodes) if dimension == "color" else _size_container(nodes) if dimension == "size" else None
if container is None: raise SkuSelectionError("未知规格维度,已停止操作。")
candidates = [n for n in nodes if _descendant(n, container) and _contained(n, container) and _choice(n) and _label(n) is not None]
return [n for n in candidates if not _labeled_ancestor(n, candidates)]
def _color_container(nodes: list[_Node]) -> _Node:
return _one([n for n in nodes if n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "androidx.recyclerview.widget.RecyclerView" and n.bounds == _COLOR_REGION], "规格面板颜色容器不唯一,已停止操作。")
def _size_container(nodes: list[_Node]) -> _Node:
label = _one([n for n in nodes if n.text == _SIZE and n.bounds == _SIZE_LABEL and _readonly(n)], "规格面板尺码标签不唯一,已停止操作。")
header = label.parent
if header is None or header.element.get("package") != PDD_PACKAGE or header.element.get("class") != "android.widget.LinearLayout" or header.bounds != _SIZE_HEADER or header.parent is None:
raise SkuSelectionError("规格面板尺码标题容器不符合已取证结构,已停止操作。")
return _one([n for n in nodes if n.parent is header.parent and n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "android.widget.LinearLayout" and n.bounds == _SIZE_OPTIONS], "规格面板尺码选项容器不唯一,已停止操作。")
def _label(node: _Node) -> str | None:
values = {value for value in (node.text, node.desc) if value}
return values.pop() if len(values) == 1 else None
def _labeled_ancestor(node: _Node, candidates: list[_Node]) -> bool:
ids, parent = {id(n.element) for n in candidates}, node.parent
while parent is not None:
if id(parent.element) in ids and _label(parent) is not None: return True
parent = parent.parent
return False
def _descendant(node: _Node, ancestor: _Node) -> bool:
parent = node.parent
while parent is not None:
if parent.element is ancestor.element: return True
parent = parent.parent
return False
def _contained(node: _Node, container: _Node) -> bool:
left, top, right, bottom = _action_bounds(node.bounds)
outer_left, outer_top, outer_right, outer_bottom = _action_bounds(container.bounds)
return outer_left <= left < right <= outer_right and outer_top <= top < bottom <= outer_bottom
def _clickable_ancestor(node: _Node) -> bool:
parent = node.parent
while parent is not None:
if parent.element.get("clickable") == "true": return True
parent = parent.parent
return False
def _readonly(node: _Node) -> bool:
return node.element.get("package") == PDD_PACKAGE and node.element.get("class") == "android.widget.TextView" and node.element.get("clickable") == "false" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true"
def _live(node: _Node) -> bool:
return node.element.get("package") == PDD_PACKAGE and node.element.get("clickable") == "true" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" and bool(node.bounds)
def _choice(node: _Node) -> bool:
return _live(node) and node.element.get("class") == "android.view.ViewGroup" and node.element.get("selected") in {"true", "false"}
def _eligible_entries(nodes: list[_Node]) -> list[_Node]:
# 入口文本本身不可点击:必须逐层证明它仍位于已取证的唯一可点击祖先中,但动作坐标继续
# 使用文本子节点的窄 bounds,避免把同一祖先内未知区域变成坐标兜底。“免拼购买”等底部
# 容器既不属于这条祖先链,也绝不能作为替代入口。
if any(node.bounds == _PRICE_PARENT for node in nodes): return []
entry_labels = [
node for node in nodes
if node.text == _ENTRY
and node.element.get("package") == PDD_PACKAGE
and node.element.get("class") == "android.widget.TextView"
]
if len(entry_labels) != 1:
return entry_labels
action_ancestors = [
node for node in nodes
if _exact_entry_node(node, "android.view.ViewGroup", _ENTRY_ACTION_BOUNDS, "true")
]
if len(action_ancestors) != 1:
return []
action_ancestor = action_ancestors[0]
entries: list[_Node] = []
for node in entry_labels:
if not _exact_entry_node(node, "android.widget.TextView", _ENTRY_TEXT_BOUNDS, "false"):
return []
inner = node.parent
switcher = inner.parent if inner is not None else None
frame = switcher.parent if switcher is not None else None
ancestor = frame.parent if frame is not None else None
if (
inner is not None
and _exact_entry_node(inner, "android.view.ViewGroup", _ENTRY_INNER_BOUNDS, "false")
and switcher is not None
and _exact_entry_node(switcher, "android.widget.ViewSwitcher", _ENTRY_INNER_BOUNDS, "false")
and frame is not None
and _exact_entry_node(frame, "android.widget.FrameLayout", _ENTRY_INNER_BOUNDS, "false")
and ancestor is action_ancestor
):
entries.append(node)
return entries
def _exact_entry_node(node: _Node, class_name: str, bounds: str, clickable: str) -> bool:
return (
node.element.get("package") == PDD_PACKAGE
and node.element.get("class") == class_name
and node.bounds == bounds
and node.element.get("clickable") == clickable
and node.element.get("enabled") == "true"
and node.element.get("visible-to-user") == "true"
)
def _action_bounds(bounds: str) -> tuple[int, int, int, int]:
match = _BOUNDS.fullmatch(bounds)
if match is None: raise SkuSelectionError("规格节点坐标格式无效,已停止操作。")
left, top, right, bottom = (int(item) for item in match.groups())
if not (0 <= left < right <= _W and 0 <= top < bottom <= _H):
raise SkuSelectionError("规格节点坐标不在已取证屏幕范围内,已停止操作。")
return left, top, right, bottom
def _one(nodes: list[_Node], message: str) -> _Node:
if len(nodes) != 1: raise SkuSelectionError(message)
return nodes[0]
@@ -1,348 +0,0 @@
"""T-103 真机运行边界:窄适配器、原始截图和无页面正文的摘要。"""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from hashlib import sha256
import json
from math import isfinite
import os
from pathlib import Path
import shutil
from time import monotonic
from typing import Any
from uuid import uuid4
from PIL import Image, UnidentifiedImageError
from adbutils.errors import AdbTimeout
from uiautomator2.exceptions import HTTPTimeoutError
from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file
from .product_open import EXPECTED_PDD_VERSION
from .product_url import ProductUrl, parse_product_url
from .sku_selection import (
EXPECTED_GOODS_ID,
EXPECTED_UNIT_PRICE,
SkuPanelDevice,
SkuSelectionError,
SkuSelectionFlow,
_action_bounds,
resolve_task_selection,
)
EXPECTED_DEVICE_MODEL = "PKG110"
EXPECTED_ANDROID_VERSION = "16"
EXPECTED_SCREEN_SIZE = (1080, 2376)
class SkuSelectionRunError(RuntimeError):
"""T-103 运行未完整完成;错误文本不携带设备或页面原文。"""
class SkuSelectionRunTimeoutError(SkuSelectionRunError):
"""设备 RPC 或操作超时。"""
class SkuSelectionScreenshotError(SkuSelectionRunError):
"""原始截图无法作为完整 PNG 原子发布。"""
class SkuSelectionUnexpectedPriceError(SkuSelectionRunError):
"""取证面板现价不是本任务已确认值。"""
class SkuSelectionDeviceAdapterError(SkuSelectionRunError):
"""第三方设备接口失败的脱敏映射。"""
@dataclass(frozen=True)
class SkuSelectionRunResult:
"""已发布的截图和无页面正文 manifest 摘要。"""
output_directory: Path
screenshot_path: Path
manifest_path: Path
unit_price: str
class UiautomatorSkuPanelAdapter(SkuPanelDevice):
"""把 uiautomator2 缩为 T-103 所需的读取与三种命名操作。
``tap_sku_entry``、``tap_sku_option`` 和 ``leave_sku_panel`` 是仅有的状态改变方法;
坐标由 Flow 和本类双重检查后才计算中心点,每次调用只执行一次底层动作。
"""
def __init__(self, device: Any, timeout_seconds: float) -> None:
if not _is_positive_finite(timeout_seconds):
raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
self._device = device
self._timeout_seconds = timeout_seconds
self._entry_was_tapped = False
self._left_panel = False
@property
def entry_was_tapped(self) -> bool:
"""仅供运行器决定故障后的单次尽力返回,不是页面操作。"""
return self._entry_was_tapped
@property
def left_panel(self) -> bool:
return self._left_panel
def app_info(self, package_name: str) -> dict[str, Any]:
value = self._call("app_info", package_name)
if not isinstance(value, dict):
raise SkuSelectionDeviceAdapterError("无法读取应用版本,已停止操作。")
return value
def app_current(self) -> dict[str, Any]:
value = self._call("app_current")
if not isinstance(value, dict):
raise SkuSelectionDeviceAdapterError("无法读取前台应用,已停止操作。")
return value
def dump_window_hierarchy(self) -> str:
value = self._call("jsonrpc_call", "dumpWindowHierarchy", [False, 50], timeout=self._timeout_seconds)
if not isinstance(value, str):
raise SkuSelectionDeviceAdapterError("节点树读取失败,已停止操作。")
return value
def tap_sku_entry(self, bounds: str) -> None:
# 超时也可能表示底层事件已经送达;必须先封存 attempt,后续绝不重试该入口。
self._entry_was_tapped = True
self._tap_bounds_once(bounds)
def tap_sku_option(self, bounds: str) -> None:
self._tap_bounds_once(bounds)
def leave_sku_panel(self) -> None:
if self._left_panel:
raise SkuSelectionDeviceAdapterError("规格面板已经执行过返回,已停止操作。")
# 底层调用即使报错也可能已把返回事件送达;先封存本次机会,finally 不得再次返回。
self._left_panel = True
self._call("jsonrpc_call", "pressKey", ["back"], timeout=self._timeout_seconds)
def capture_screenshot(self) -> str:
value = self._call("jsonrpc_call", "takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds)
if not isinstance(value, str):
raise SkuSelectionScreenshotError("规格面板原始截图读取失败,未发布任何证据产物。")
return value
def display_size(self) -> tuple[int, int]:
value = self._call("window_size")
if not isinstance(value, tuple) or len(value) != 2 or any(not isinstance(item, int) for item in value):
raise SkuSelectionDeviceAdapterError("无法读取屏幕坐标空间,已停止操作。")
return value
def _tap_bounds_once(self, bounds: str) -> None:
left, top, right, bottom = _action_bounds(bounds)
center_x = left + (right - left) // 2
center_y = top + (bottom - top) // 2
self._call("jsonrpc_call", "click", [center_x, center_y], timeout=self._timeout_seconds)
def _call(self, method: str, *args: Any, **kwargs: Any) -> Any:
try:
operation = getattr(self._device, method)
return operation(*args, **kwargs)
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
raise SkuSelectionRunTimeoutError("规格面板设备操作超时,已停止操作。") from error
except SkuSelectionRunError:
raise
except Exception as error:
raise SkuSelectionDeviceAdapterError("规格面板设备操作失败,已停止操作。") from error
class SkuSelectionRunner:
"""只运行 T-103 目标规格恢复、价格确认、原始截图和一次安全退出。"""
def __init__(
self,
adb_client: AdbClient,
connector: Callable[[str], Any],
timeout_seconds: float,
monotonic_clock: Callable[[], float] = monotonic,
) -> None:
if not _is_positive_finite(timeout_seconds):
raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
self._adb_client = adb_client
self._connector = connector
self._timeout_seconds = timeout_seconds
self._monotonic_clock = monotonic_clock
def run(
self,
serial: str,
product_url: str,
task_color: str,
task_size: str,
output_directory: Path,
) -> SkuSelectionRunResult:
link = parse_product_url(product_url)
if link.goods_id != EXPECTED_GOODS_ID:
raise SkuSelectionRunError("商品不是 T-103 已取证目标,已停止操作。")
selection = resolve_task_selection(task_color, task_size)
target = Path(output_directory)
_validate_new_target(target)
adapter: UiautomatorSkuPanelAdapter | None = None
flow: SkuSelectionFlow | None = None
staging = _prepare_staging(target)
deadline = self._monotonic_clock() + self._timeout_seconds
try:
inspection = self._adb_client.inspect(serial)
_require_expected_device(inspection)
adapter = UiautomatorSkuPanelAdapter(self._connector(serial), self._timeout_seconds)
_require_expected_version(adapter.app_info(PDD_PACKAGE))
if adapter.display_size() != EXPECTED_SCREEN_SIZE:
raise SkuSelectionRunError("设备不是已取证的竖屏坐标空间,已停止操作。")
pre_intent_hierarchy = adapter.dump_window_hierarchy()
# 固定 ACTION_VIEW、固定 PDD package 和 canonical goods_id;不接受任意 URL 或 shell。
self._adb_client.start_pdd_view_intent(serial, link.goods_id)
remaining = deadline - self._monotonic_clock()
if remaining <= 0:
raise SkuSelectionRunTimeoutError("等待规格入口超时,未执行点击。")
flow = SkuSelectionFlow(adapter, entry_wait_timeout_seconds=remaining)
flow.open_sku_panel(link.canonical_url, pre_intent_hierarchy)
flow.select_sku_options(selection)
unit_price = flow.verify_target_selection_and_read_price(selection)
if unit_price != EXPECTED_UNIT_PRICE:
raise SkuSelectionUnexpectedPriceError("规格面板现价不是本任务已确认值,已停止操作。")
screenshot_path = staging / "screenshot.png"
try:
_save_base64_screenshot(adapter.capture_screenshot(), screenshot_path)
_require_screenshot_size(screenshot_path)
except SkuSelectionRunError:
raise
except Exception as error:
raise SkuSelectionScreenshotError("规格面板原始截图保存失败,未发布任何证据产物。") from error
manifest_path = staging / "manifest.json"
# 截图可能落在动态页面切换边界;发布前必须用一棵更新节点树同时重证两维和现价。
final_price = flow.verify_target_selection_and_read_price(selection)
if final_price != EXPECTED_UNIT_PRICE:
raise SkuSelectionUnexpectedPriceError("截图后规格面板现价不是本任务已确认值,已停止操作。")
# 正常路径仍经 Flow 做最后一次前台和面板判定;返回操作只发生一次。
flow.exit_sku_panel_safely()
manifest_path.write_text(
json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
# Windows 的 rename 不替换既有目标;并发创建 target 时保留其内容并把本次运行判失败。
os.rename(staging, target)
staging = None
except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError):
_clean_staging(staging)
raise
except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
_clean_staging(staging)
raise SkuSelectionRunTimeoutError("规格面板运行超时,未发布任何证据产物。") from error
except OSError as error:
_clean_staging(staging)
raise SkuSelectionRunError("规格面板证据目录无法创建或发布,未发布任何证据产物。") from error
except Exception as error:
_clean_staging(staging)
raise SkuSelectionRunError("规格面板运行未完成,未发布任何证据产物。") from error
finally:
# 失败路径只能复用 Flow 的版本、前台和面板证明;证明不了便停止,绝不盲目返回。
if flow is not None and adapter is not None and adapter.entry_was_tapped and not adapter.left_panel:
try:
flow.reconcile_pending_action()
flow.exit_sku_panel_safely()
except (SkuSelectionRunError, SkuSelectionError):
pass
return SkuSelectionRunResult(
output_directory=target,
screenshot_path=target / "screenshot.png",
manifest_path=target / "manifest.json",
unit_price=EXPECTED_UNIT_PRICE,
)
def _is_positive_finite(value: object) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
def _validate_new_target(target: Path) -> None:
if target.exists():
raise SkuSelectionRunError("输出目录已存在;为防止覆盖旧证据,已停止操作。")
if not target.name:
raise SkuSelectionRunError("输出目录必须是明确的新目录。")
def _prepare_staging(target: Path) -> Path:
staging: Path | None = None
try:
target.parent.mkdir(parents=True, exist_ok=True)
staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
staging.mkdir()
probe = staging / ".write-probe"
probe.write_bytes(b"ok")
probe.unlink()
return staging
except OSError as error:
_clean_staging(staging)
raise SkuSelectionRunError("输出目录不可写,已停止操作。") from error
def _clean_staging(staging: Path | None) -> None:
if staging is not None and staging.exists():
shutil.rmtree(staging)
def _require_expected_version(app_info: object) -> str:
version = (app_info.get("versionName") or app_info.get("version_name")) if isinstance(app_info, dict) else None
if version != EXPECTED_PDD_VERSION:
raise SkuSelectionRunError("拼多多版本与已取证版本不一致,已停止操作。")
return version
def _require_expected_device(inspection: DeviceInspection) -> None:
if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION:
raise SkuSelectionRunError("设备型号或 Android 版本不是已取证组合,已停止操作。")
def _require_screenshot_size(screenshot_path: Path) -> None:
try:
with Image.open(screenshot_path) as image:
image.load()
if image.size != EXPECTED_SCREEN_SIZE:
raise SkuSelectionScreenshotError("原始截图坐标空间不是已取证尺寸,未发布任何证据产物。")
except SkuSelectionRunError:
raise
except (UnidentifiedImageError, OSError) as error:
raise SkuSelectionScreenshotError("原始截图无效,未发布任何证据产物。") from error
def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, screenshot_path: Path, task_color: str, task_size: str) -> dict[str, Any]:
"""仅写可审计摘要;原始 serial、节点树、页面文案和实际截图内容均不写入 manifest。"""
return {
"schema_version": 1,
"captured_at": datetime.now(UTC).isoformat(),
"operation": "t103-sku-selection",
"product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
"target_selection": {"color": task_color, "size": task_size},
"unit_price": EXPECTED_UNIT_PRICE,
"selection_status": "restored",
"panel_status": "verified",
"safe_exit": "completed",
"page_identity": "human_review_required",
"channel": "wifi" if ":" in serial else "usb",
"serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),
"device": {
"model": inspection.model,
"android_version": inspection.android_version,
"pdd_package": PDD_PACKAGE,
"pdd_version": EXPECTED_PDD_VERSION,
},
"artifacts": [{"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)}],
}
@@ -1,14 +0,0 @@
<hierarchy>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[0,1256][1080,1355]" clickable="true" enabled="true" visible-to-user="true">
<node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout" bounds="[712,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true">
<node package="com.xunmeng.pinduoduo" class="android.widget.ViewSwitcher" bounds="[712,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true">
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[712,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true">
<node text="快要抢光" package="com.xunmeng.pinduoduo" class="android.widget.TextView" bounds="[900,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true" />
</node>
</node>
</node>
</node>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[446,2166][1080,2328]" clickable="true" enabled="true" visible-to-user="true">
<node text="免拼购买" package="com.xunmeng.pinduoduo" class="android.widget.TextView" bounds="[688,2256][856,2305]" clickable="false" enabled="true" visible-to-user="true" />
</node>
</hierarchy>
@@ -1,26 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<hierarchy>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[0,474][1080,2328]">
<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">
<node text="快卖完 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[396,503][712,570]" />
<node text="¥29.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[730,503][895,570]" />
</node>
<node text="已选: 黑色 CHA (纯棉) M(建议100-115)" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[396,654][1053,716]" />
<node package="com.xunmeng.pinduoduo" class="androidx.recyclerview.widget.RecyclerView" bounds="[36,1000][1080,1631]">
<node content-desc="黑色 CHA (纯棉)" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]" />
<node content-desc="粉红" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="false" clickable="true" enabled="true" visible-to-user="true" bounds="[456,1000][690,1172]" />
</node>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[36,1637][1044,2045]">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[36,1637][1044,1718]">
<node text="尺码" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[36,1654][114,1700]" />
</node>
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[36,1730][1044,2045]">
<node text="M(建议100-115)" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[439,1730][831,1815]" />
<node text="L(建议115-130)" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="false" clickable="true" enabled="true" visible-to-user="true" bounds="[840,1730][1044,1815]" />
</node>
</node>
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" clickable="true" enabled="true" visible-to-user="true" bounds="[357,2181][722,2328]">
<node text="提交订单 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[369,2225][710,2284]" />
</node>
</node>
</hierarchy>
-689
View File
@@ -1,689 +0,0 @@
from __future__ import annotations
import ast
import base64
from contextlib import redirect_stderr
from io import BytesIO
import importlib.util
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
from xml.etree import ElementTree
from PIL import Image
import cmbuyer_client.pdd as pdd
import cmbuyer_client.pdd.sku_selection_runner as runner_module
from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection
from cmbuyer_client.pdd.sku_selection_runner import (
SkuSelectionDeviceAdapterError,
SkuSelectionRunError,
SkuSelectionScreenshotError,
UiautomatorSkuPanelAdapter,
)
_FIXTURE = Path(__file__).with_name("fixtures") / "sku_panel_8_17_0.xml"
_ENTRY_FIXTURE = Path(__file__).with_name("fixtures") / "product_entry_8_17_0.xml"
_TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375"
_TASK_COLOR = "黑色CHA(纯棉)"
_TASK_SIZE = "M(建议100-115)"
_PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")
def _png_base64() -> str:
image = Image.new("RGB", (1080, 2376), "white")
raw = BytesIO()
image.save(raw, format="PNG")
return base64.b64encode(raw.getvalue()).decode("ascii")
class _RawDevice:
def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
self.hierarchy = hierarchy
self.panel_hierarchy = _FIXTURE.read_text(encoding="utf-8")
self.version = "8.17.0"
self.package = "com.xunmeng.pinduoduo"
self.screenshot = _png_base64() if screenshot is None else screenshot
self.calls: list[tuple[object, ...]] = []
self.fail_color_readback = False
def app_info(self, package_name: str) -> dict[str, str]:
self.calls.append(("app_info", package_name))
return {"versionName": self.version}
def app_current(self) -> dict[str, str]:
self.calls.append(("app_current",))
return {"package": self.package}
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
self.calls.append(("jsonrpc", method, params, timeout))
if method == "dumpWindowHierarchy":
return self.hierarchy
if method == "takeScreenshot":
return self.screenshot
if method == "pressKey":
self.hierarchy = "<hierarchy />"
return ""
if method == "click":
if not isinstance(params, list) or len(params) != 2:
raise AssertionError(params)
self._apply_tap(int(params[0]), int(params[1]))
return ""
raise AssertionError(method)
def _apply_tap(self, x: int, y: int) -> None:
if "快要抢光" in self.hierarchy and "[396,498][895,570]" not in self.hierarchy:
self.hierarchy = self.panel_hierarchy
return
root = ElementTree.fromstring(self.hierarchy)
target = next(node for node in root.iter("node") if _center(node.get("bounds", "")) == (x, y))
color = target.get("bounds", "").endswith("][438,1172]")
for node in root.iter("node"):
if node.get("selected") is not None and ((color and ",1000]" in node.get("bounds", "")) or (not color and ",1730]" in node.get("bounds", ""))):
node.set("selected", "false")
if color and self.fail_color_readback:
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
else:
target.set("selected", "true")
self.hierarchy = ElementTree.tostring(root, encoding="unicode")
def window_size(self) -> tuple[int, int]:
self.calls.append(("window_size",))
return 1080, 2376
def select_alternates(self) -> None:
root = ElementTree.fromstring(self.panel_hierarchy)
for node in root.iter("node"):
if node.get("selected") is not None:
node.set("selected", "false")
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
next(node for node in root.iter("node") if node.get("text") == "L(建议115-130)").set("selected", "true")
self.panel_hierarchy = ElementTree.tostring(root, encoding="unicode")
if self.hierarchy != _PRODUCT_PAGE:
self.hierarchy = self.panel_hierarchy
def _center(bounds: str) -> tuple[int, int]:
left_top, right_bottom = bounds.split("][")
left, top = (int(value) for value in left_top.removeprefix("[").split(","))
right, bottom = (int(value) for value in right_bottom.removesuffix("]").split(","))
return left + (right - left) // 2, top + (bottom - top) // 2
def _entry_chain(root: ElementTree.Element) -> list[ElementTree.Element]:
parents = {child: parent for parent in root.iter() for child in parent}
child = next(node for node in root.iter("node") if node.get("text") == "快要抢光")
chain = [child]
for _ in range(4):
chain.append(parents[chain[-1]])
return chain
def _mutate_entry(depth: int, attribute: str, value: str) -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
_entry_chain(root)[depth].set(attribute, value)
return ElementTree.tostring(root, encoding="unicode")
def _without_entry() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
root.remove(_entry_chain(root)[4])
return ElementTree.tostring(root, encoding="unicode")
def _duplicate_entry() -> str:
root = ElementTree.fromstring(_PRODUCT_PAGE)
entry_root = _entry_chain(root)[4]
root.append(ElementTree.fromstring(ElementTree.tostring(entry_root, encoding="unicode")))
return ElementTree.tostring(root, encoding="unicode")
def _actions(device: _RawDevice, method: str) -> list[tuple[object, ...]]:
return [call for call in device.calls if call[0] == "jsonrpc" and call[1] == method]
def _tap_centers(device: _RawDevice) -> list[tuple[int, int]]:
return [tuple(call[2]) for call in _actions(device, "click")] # type: ignore[misc]
class _FakeAdb:
def __init__(self) -> None:
self.calls: list[tuple[object, ...]] = []
self.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "PKG110", "16")
self.on_intent: callable | None = None
def inspect(self, serial: str) -> DeviceInspection:
self.calls.append(("inspect", serial))
return self.inspection
def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
self.calls.append(("intent", serial, goods_id))
if self.on_intent is not None:
self.on_intent()
return object()
class SkuSelectionFlowTests(unittest.TestCase):
def _assert_entry_rejected_without_click(self, hierarchy: str) -> None:
now = [0.0]
device = _RawDevice(hierarchy)
flow = SkuSelectionFlow(
UiautomatorSkuPanelAdapter(device, 10),
0.01,
0.01,
lambda: now[0],
lambda seconds: now.__setitem__(0, now[0] + seconds),
)
with self.assertRaises(SkuSelectionError):
flow.open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_target_mapping_is_exact_and_success_path_restores_target(self) -> None:
device = _RawDevice()
adapter = UiautomatorSkuPanelAdapter(device, 10)
flow = SkuSelectionFlow(adapter)
flow.open_sku_panel(_TARGET_URL)
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(flow.read_sku_unit_price(), "12.88")
flow.exit_sku_panel_safely()
self.assertEqual(_tap_centers(device), [(978, 1333)])
self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
device = _RawDevice()
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
self.assertEqual(_tap_centers(device), [(978, 1333)])
def test_entry_child_and_every_ancestor_attribute_drift_never_clicks(self) -> None:
expected_clickable = ("false", "false", "false", "false", "true")
for depth in range(5):
changes = {
"package": "other.package",
"class": "android.widget.Button",
"bounds": "[1,1][2,2]",
"clickable": "true" if expected_clickable[depth] == "false" else "false",
"enabled": "false",
"visible-to-user": "false",
}
for attribute, value in changes.items():
with self.subTest(depth=depth, attribute=attribute):
self._assert_entry_rejected_without_click(_mutate_entry(depth, attribute, value))
def test_duplicate_entry_and_forbidden_sibling_entry_never_click(self) -> None:
self._assert_entry_rejected_without_click(_duplicate_entry())
self._assert_entry_rejected_without_click(_without_entry())
self._assert_entry_rejected_without_click(_mutate_entry(0, "clickable", "true"))
def test_unknown_task_or_ui_variants_are_rejected_without_action(self) -> None:
for color, size in (("黑色 CHA (纯棉)", _TASK_SIZE), (_TASK_COLOR, "M(建议100-115)"), ("黑色CHA(纯棉)", _TASK_SIZE)):
with self.subTest(color=color, size=size), self.assertRaises(SkuSelectionError):
resolve_task_selection(color, size)
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8"))
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).select_sku_options(
resolve_task_selection(_TASK_COLOR, _TASK_SIZE).__class__("粉红", "L(建议115-130)")
)
self.assertEqual(_actions(device, "click"), [])
def test_option_selected_and_container_drift_fail_closed_before_click(self) -> None:
base = _FIXTURE.read_text(encoding="utf-8")
cases = (
base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'selected="maybe" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
base.replace('bounds="[126,1000][438,1172]"', 'bounds="[1,1][20,20]"'),
base.replace('enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'enabled="false" visible-to-user="true" bounds="[126,1000][438,1172]"'),
)
for hierarchy in cases:
with self.subTest(), self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(hierarchy), 10)).select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
def test_invalid_bounds_stop_before_action(self) -> None:
for bounds in ("", "[1,2][1,3]", "[1,2][3,2]", "[0,0][1081,1]", "[0,0][1,2377]", "[a,0][1,1]"):
with self.subTest(bounds=bounds), self.assertRaises(SkuSelectionError):
_action_bounds(bounds)
device = _RawDevice(_PRODUCT_PAGE.replace("[900,1312][1056,1355]", "[0,0][1081,1]"))
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_color_readback_failure_never_attempts_second_option(self) -> None:
device = _RawDevice()
device.fail_color_readback = True
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
flow.open_sku_panel(_TARGET_URL)
device.select_alternates()
with self.assertRaises(SkuSelectionError):
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(_tap_centers(device), [(978, 1333), (282, 1086)])
def test_non_target_selection_restores_each_dimension_once(self) -> None:
device = _RawDevice()
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
flow.open_sku_panel(_TARGET_URL)
device.select_alternates()
flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
self.assertEqual(
_tap_centers(device),
[(978, 1333), (282, 1086), (635, 1772)],
)
def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None:
for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"):
with self.subTest(replacement=replacement):
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", replacement))
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", "提交订单 ¥12.88"))
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
clickable_parent = _FIXTURE.read_text(encoding="utf-8").replace(
'<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">',
'<node package="" class="android.view.ViewGroup" clickable="true" bounds="[396,498][895,570]">',
)
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(clickable_parent), 10)).read_sku_unit_price()
def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None:
forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click"}
self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
self.assertTrue(forbidden.isdisjoint(pdd.__all__))
def test_static_ast_boundary_limits_flow_runner_adapter_and_cli(self) -> None:
root = Path(__file__).resolve().parents[2]
files = (
root / "src" / "cmbuyer_client" / "pdd" / "sku_selection.py",
root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py",
root / "scripts" / "run_t103_sku_selection.py",
)
forbidden = ("quantity", "confirm", "authorization", "fence", "submit_order", "payment")
for path in files:
source = path.read_text(encoding="utf-8")
with self.subTest(path=path.name):
self.assertTrue(all(token not in source.lower() for token in forbidden))
tree = ast.parse(source)
self.assertFalse(any(isinstance(node, ast.ImportFrom) and node.module in {"selenium", "requests"} for node in ast.walk(tree)))
runner_tree = ast.parse(files[1].read_text(encoding="utf-8"))
click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"]
self.assertEqual(len(click_calls), 1)
def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
now = [0.0]
device = _RawDevice()
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), 0.01, 0.01, lambda: now[0], lambda seconds: now.__setitem__(0, now[0] + seconds))
with self.assertRaises(SkuSelectionError):
flow.open_sku_panel(_TARGET_URL, _PRODUCT_PAGE)
self.assertEqual(_actions(device, "click"), [])
def test_action_postcondition_wait_never_repeats_entry_click(self) -> None:
class NoPanelAfterEntry(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.calls.append(("jsonrpc", method, params, timeout))
return ""
return super().jsonrpc_call(method, params, timeout)
device = NoPanelAfterEntry()
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
self.assertEqual(_tap_centers(device), [(978, 1333)])
duplicate = _duplicate_entry()
device = _RawDevice(duplicate)
with self.assertRaises(SkuSelectionError):
SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
self.assertEqual(_actions(device, "click"), [])
def test_fixture_contains_no_address_phone_or_payment_credentials(self) -> None:
for fixture in (_FIXTURE, _ENTRY_FIXTURE):
content = fixture.read_text(encoding="utf-8")
with self.subTest(fixture=fixture.name):
self.assertNotRegex(content, r"1[3-9]\d{9}")
for forbidden in ("地址", "收货", "支付", "银行卡", "身份证"):
self.assertNotIn(forbidden, content)
content = _FIXTURE.read_text(encoding="utf-8")
root = ElementTree.fromstring(content)
leaf = next(node for node in root.iter("node") if node.get("text") == "提交订单 ¥12.88")
self.assertEqual(leaf.get("clickable"), "false")
self.assertEqual(leaf.get("bounds"), "[369,2225][710,2284]")
class SkuSelectionRunnerTests(unittest.TestCase):
def _runner(self, adb: _FakeAdb, device: _RawDevice) -> SkuSelectionRunner:
device.hierarchy = "<hierarchy />"
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
return SkuSelectionRunner(adb, lambda serial: device, 10)
def test_runner_atomically_publishes_screenshot_and_redacted_manifest(self) -> None:
adb = _FakeAdb()
device = _RawDevice()
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual(result.unit_price, "12.88")
manifest = result.manifest_path.read_text(encoding="utf-8")
self.assertTrue(result.screenshot_path.is_file())
self.assertNotIn("device-1", manifest)
self.assertNotIn("hierarchy", manifest)
self.assertNotIn("已选", manifest)
self.assertIn('"unit_price": "12.88"', manifest)
self.assertIn('"selection_status": "restored"', manifest)
self.assertIn('"panel_status": "verified"', manifest)
self.assertIn('"safe_exit": "completed"', manifest)
self.assertFalse((target / "hierarchy.xml").exists())
self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
original_rename = runner_module.os.rename
def create_target_then_rename(source: str | Path, destination: str | Path) -> None:
Path(destination).mkdir()
(Path(destination) / "sentinel").write_text("keep", encoding="utf-8")
original_rename(source, destination)
with patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual((target / "sentinel").read_text(encoding="utf-8"), "keep")
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
with TemporaryDirectory() as temporary:
target = Path(temporary) / "result"
with self.assertRaises(SkuSelectionScreenshotError):
self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
target = Path(temporary) / "write-failure"
with patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")):
with self.assertRaises(SkuSelectionScreenshotError):
self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertEqual(list(Path(temporary).glob(".write-failure.staging-*")), [])
adb = _FakeAdb()
device = _RawDevice()
target.mkdir()
sentinel = target / "keep"
sentinel.write_text("keep", encoding="utf-8")
with self.assertRaises(SkuSelectionRunError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertEqual(adb.calls, [])
self.assertEqual(device.calls, [])
self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
def test_device_screen_and_output_preflight_fail_before_any_click(self) -> None:
with TemporaryDirectory() as temporary:
adb = _FakeAdb()
adb.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "wrong", "16")
device = _RawDevice()
with self.assertRaises(SkuSelectionRunError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(device.calls, [])
class WrongScreenDevice(_RawDevice):
def window_size(self) -> tuple[int, int]:
return 1080, 1920
device = WrongScreenDevice()
with self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "screen")
self.assertEqual(_actions(device, "click"), [])
parent_file = Path(temporary) / "not-a-directory"
parent_file.write_text("x", encoding="utf-8")
adb = _FakeAdb()
device = _RawDevice()
with self.assertRaises(SkuSelectionRunError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, parent_file / "result")
self.assertEqual(adb.calls, [])
self.assertEqual(device.calls, [])
def test_small_but_valid_png_is_not_accepted(self) -> None:
image = Image.new("RGB", (1, 1), "white")
raw = BytesIO(); image.save(raw, format="PNG")
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionScreenshotError):
self._runner(_FakeAdb(), _RawDevice(screenshot=base64.b64encode(raw.getvalue()).decode("ascii"))).run(
"device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
)
def test_failure_after_entry_attempts_one_safe_exit_and_hides_device_detail(self) -> None:
adb = _FakeAdb()
device = _RawDevice()
device.fail_color_readback = True
device.select_alternates()
with TemporaryDirectory() as temporary:
with self.assertRaises(SkuSelectionError):
self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(_actions(device, "pressKey"), [])
class FailingRawDevice(_RawDevice):
def app_info(self, package_name: str) -> dict[str, str]:
raise RuntimeError("device-1 <xml>private</xml>")
with self.assertRaises(SkuSelectionDeviceAdapterError) as raised:
UiautomatorSkuPanelAdapter(FailingRawDevice(), 10).app_info("com.xunmeng.pinduoduo")
self.assertNotIn("device-1", str(raised.exception))
self.assertNotIn("private", str(raised.exception))
def test_unverified_failure_never_sends_blind_back(self) -> None:
class InvalidAfterOptionDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
value = super().jsonrpc_call(method, params, timeout)
if method == "click" and "[396,498][895,570]" in self.hierarchy:
self.hierarchy = "<hierarchy />"
return value
device = InvalidAfterOptionDevice()
device.select_alternates()
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(_actions(device, "pressKey"), [])
def test_adapter_timeout_is_mapped_without_third_party_detail(self) -> None:
class TimeoutRawDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
raise TimeoutError("device-1 <hierarchy>private</hierarchy>")
with self.assertRaises(SkuSelectionRunError) as raised:
UiautomatorSkuPanelAdapter(TimeoutRawDevice(), 10).dump_window_hierarchy()
self.assertNotIn("device-1", str(raised.exception))
self.assertNotIn("private", str(raised.exception))
def test_entry_attempt_is_recorded_before_unconfirmed_click_and_not_retried(self) -> None:
class TimeoutTapDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
self.calls.append(("jsonrpc", method, params, timeout))
if method == "click":
raise TimeoutError("device detail")
return super().jsonrpc_call(method, params, timeout)
adapter = UiautomatorSkuPanelAdapter(TimeoutTapDevice(), 10)
with self.assertRaises(SkuSelectionRunError):
adapter.tap_sku_entry("[900,1312][1056,1355]")
self.assertTrue(adapter.entry_was_tapped)
self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [978, 1333], 10)])
def test_entry_stability_interruptions_never_click(self) -> None:
now = [0.0]
class SequenceDevice(_RawDevice):
def __init__(self) -> None:
super().__init__(); self.frames = [_PRODUCT_PAGE, "<hierarchy />", _PRODUCT_PAGE]
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "dumpWindowHierarchy" and self.frames:
self.hierarchy = self.frames.pop(0)
return super().jsonrpc_call(method, params, timeout)
device = SequenceDevice()
flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), .02, .01, lambda: now[0], lambda x: now.__setitem__(0, now[0] + x))
with self.assertRaises(SkuSelectionError): flow.open_sku_panel(_TARGET_URL, "<hierarchy />")
self.assertEqual(_actions(device, "click"), [])
def test_screenshot_drift_and_foreground_drift_publish_nothing_and_never_back(self) -> None:
for drift in ("color", "size", "price"):
class DriftDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
value = super().jsonrpc_call(method, params, timeout)
if method == "takeScreenshot":
if drift == "price":
self.hierarchy = self.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88")
else:
root = ElementTree.fromstring(self.hierarchy)
if drift == "color":
for node in root.iter("node"):
if node.get("selected") is not None and ",1000]" in node.get("bounds", ""):
node.set("selected", "false")
next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
else:
for node in root.iter("node"):
if node.get("selected") is not None and ",1730]" in node.get("bounds", ""):
node.set("selected", "false")
next(node for node in root.iter("node") if node.get("text") == "L(建议115-130)").set("selected", "true")
self.hierarchy = ElementTree.tostring(root, encoding="unicode")
return value
with self.subTest(drift=drift), TemporaryDirectory() as temporary:
target = Path(temporary) / "out"
with self.assertRaises((SkuSelectionError, SkuSelectionRunError)):
self._runner(_FakeAdb(), DriftDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertFalse((target / "manifest.json").exists())
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
device = _RawDevice(); device.select_alternates()
device.package = "other"
with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).exit_sku_panel_safely()
self.assertEqual(_actions(device, "pressKey"), [])
def test_screenshot_then_foreground_drift_publishes_nothing_and_never_back(self) -> None:
class ForegroundDriftDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
value = super().jsonrpc_call(method, params, timeout)
if method == "takeScreenshot": self.package = "other"
return value
device = ForegroundDriftDevice()
with TemporaryDirectory() as temporary:
target = Path(temporary) / "out"
with self.assertRaises(SkuSelectionError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
self.assertFalse(target.exists())
self.assertFalse((target / "manifest.json").exists())
self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
self.assertEqual(_actions(device, "pressKey"), [])
def test_option_timeout_reconciliation_controls_back_once(self) -> None:
class OptionTimeoutDevice(_RawDevice):
def __init__(self, delivered: bool) -> None:
super().__init__(); self.delivered = delivered; self.clicks = 0
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click":
self.clicks += 1
if self.clicks == 2:
if self.delivered: super().jsonrpc_call(method, params, timeout)
else: self.calls.append(("jsonrpc", method, params, timeout))
raise TimeoutError("uncertain option")
return super().jsonrpc_call(method, params, timeout)
for delivered, expected_back in ((False, 0), (True, 1)):
with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
device = OptionTimeoutDevice(delivered); device.select_alternates()
adb = _FakeAdb(); device.hierarchy = "<hierarchy />"
adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
runner = SkuSelectionRunner(adb, lambda serial: device, .03)
with self.assertRaises(SkuSelectionRunError):
runner.run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
self.assertEqual(len(_actions(device, "click")), 2)
self.assertEqual(len(_actions(device, "pressKey")), expected_back)
def test_back_timeout_is_never_retried(self) -> None:
class BackTimeoutDevice(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "pressKey":
super().jsonrpc_call(method, params, timeout)
raise TimeoutError("back uncertain")
return super().jsonrpc_call(method, params, timeout)
device = BackTimeoutDevice()
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
self.assertEqual(len(_actions(device, "pressKey")), 1)
def test_entry_click_timeout_reconciles_only_through_verified_flow_exit(self) -> None:
class DeliveredThenTimeout(_RawDevice):
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
if method == "click" and self.hierarchy != _FIXTURE.read_text(encoding="utf-8"):
super().jsonrpc_call(method, params, timeout)
raise TimeoutError("delivery uncertain")
return super().jsonrpc_call(method, params, timeout)
device = DeliveredThenTimeout()
with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
self.assertEqual(len(_actions(device, "click")), 1)
self.assertEqual(len(_actions(device, "pressKey")), 1)
class SkuSelectionCliTests(unittest.TestCase):
def test_cli_accepts_only_target_url_and_task_values(self) -> None:
script = _load_runner_script()
valid = {
"serial": "device-1",
"url": _TARGET_URL,
"color": _TASK_COLOR,
"size": _TASK_SIZE,
"output_dir": Path("evidence"),
"timeout": 10.0,
"adb": "adb",
}
script.validate_arguments(type("Arguments", (), valid)())
for field, value in (("serial", ""), ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=1"), ("color", "黑色 CHA (纯棉)"), ("size", "M(建议100-115)"), ("timeout", 0), ("timeout", float("inf"))):
with self.subTest(field=field, value=value), self.assertRaises((ValueError, SkuSelectionError)):
script.validate_arguments(type("Arguments", (), valid | {field: value})())
def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
script = _load_runner_script()
class FlowFailingRunner:
def __init__(self, *args: object, **kwargs: object) -> None: pass
def run(self, *args: object, **kwargs: object) -> object:
raise SkuSelectionError("<hierarchy>page-body</hierarchy>")
stderr = BytesIO()
# TextIOWrapper keeps the assertion independent from host console encoding.
import io
text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
status = script.main([
"--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
"--size", _TASK_SIZE, "--output-dir", "evidence",
])
text_stderr.flush()
output = stderr.getvalue().decode("utf-8")
self.assertEqual(status, 1)
self.assertNotIn("Traceback", output)
self.assertNotIn("page-body", output)
def _load_runner_script() -> object:
path = Path(__file__).resolve().parents[2] / "scripts" / "run_t103_sku_selection.py"
specification = importlib.util.spec_from_file_location("run_t103_sku_selection_test", path)
if specification is None or specification.loader is None:
raise RuntimeError("无法加载 T-103 运行脚本。")
module = importlib.util.module_from_spec(specification)
specification.loader.exec_module(module)
return module
if __name__ == "__main__":
unittest.main()
+1 -10
View File
@@ -17,7 +17,6 @@ write_paths:
- client/tests/pdd/**
- client/tests/device/**
- client/scripts/capture_sku_panel_spike.py
- client/scripts/run_t103_sku_selection.py
- client/scripts/sanitize_sku_panel_evidence.py
- docs/02-requirements.md
- docs/03-tech-stack.md
@@ -26,7 +25,7 @@ write_paths:
- docs/current-state.md
---
<!-- BEGIN VIKUNJA EXPORT id=23 synced=2026-08-04T10:01:59Z sha256=ed26c948a564efd44a1f3d3336dd4c383cc42acf567357c5c8fedb32d2459a0e -->
<!-- BEGIN VIKUNJA EXPORT id=23 synced=2026-08-04T08:24:57Z sha256=19a0864d83c5e41e57a29002b69f7767b50429dcf0fbace02f8303f92b71bac6 -->
## 问题 / 背景
T-102 已证明 canonical 链接可进入目标商品。T-103 在 PKG110 / Android 16 / 拼多多 8.17.0、goods_id `937122477375` 上确认:规格面板由详情页精确唯一的“快要抢光”打开;T-110 已把该证据/版本绑定入口批准为受控导航。面板刚打开时目标颜色“黑色CHA(纯棉)”和尺码“M(建议100-115)”自动选中。
@@ -174,14 +173,6 @@ T-103 sanitizer v2 坐标修正与主审:提交 44c027a 将 screenshot space
### 2026-08-04T08:24:12Z · ila
2026-08-04:T-111 单趟契约主审通过,T-103 解除架构阻塞。恢复后只实现 SkuSelectionFlow:受控入口、维度内精确选择、选中态读回、SKU 当前价唯一读取、原始截图和安全退出;不得包含数量、确认页、授权、提交围栏、提交订单或支付。客户端工作树现有未提交 v6 遮罩测试残留必须先撤销到已提交 v5 基线,再开始新实现。needs_device=true,离线实现后仍等待人工真机验收。
### 2026-08-04T09:50:25Z · ila
2026-08-04 离线实现与主审通过:提交 b49a9b4 实现仅限已取证 PDD 8.17.0 / goods_id 937122477375 的受控规格入口、颜色/尺码精确恢复、SKU 当前价读取、本机原始截图和单次安全退出;不包含数量、确认页、授权、提交围栏、提交订单或付款能力。独立审计 PASS,合入主分支 5f650f1;完整 init、109 项测试、Go vet/build、compileall、上下文校验及 diff-check 全部通过。needs_device=true,任务继续 DOING,等待项目所有者先把已记忆规格改为非目标值并返回 PDD 首页后运行真机验收。
### 2026-08-04T10:01:50Z · ila
2026-08-04 T-103 入口父容器追加真机证据:只读取证目录 C:\Users\ila20\AppData\Local\cmbuyer\artifacts\T-103\entry-parent-evidence-937122477375-20260804-175737;截图 screenshot.png,XML hierarchy.xml,manifest.json。设备 PKG110 / Android 16 / 拼多多 8.17.0 / Wi-Fi,goods_id 937122477375。项目所有者人工确认截图为目标商品详情页,且“快要抢光 12.88”与“免拼购买”的位置和手机当前画面一致。只读结构核对显示:精确“快要抢光”文本节点本身不可点击,但位于唯一、可见、启用、可点击的 PDD ViewGroup 祖先内;“免拼购买”属于另一底部可点击容器。后续只允许把入口判据收紧调整为“精确唯一快要抢光子节点 + 证据绑定唯一可点击祖先”,不得允许或点击“免拼购买”,必须先补 fixture/反例/超时不重试测试再真机运行。
<!-- END VIKUNJA EXPORT -->
## 边界
+1
View File
@@ -12,6 +12,7 @@ needs_device: false
needs_human_review: false
write_paths:
- docs/tasks/T-203.md
- admin/internal/auth/**
- admin/internal/tasks/**
- admin/internal/server/**
- admin/internal/transport/webui/**