feat: 持久化采集记录去重 + Goods系统批量导入
CI / Go (api) (pull_request) Successful in 55s
CI / Python (ingestion) (pull_request) Failing after 31s
CI / Migrations (postgres) (pull_request) Failing after 32s

- collected.txt 跨文件/跨次记录已查询条码,避免重复采集
- Web UI 显示历史已采集条码计数
- Goods API 新增 POST /api/import/bypos 批量导入端点(含upsert)
- 采集器 Web UI 新增一键导入按钮(登录+上传JSONL)
- Windows exe 重新编译

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
ceyhandagdas51272
2026-06-24 04:40:26 +00:00
parent cbb0968256
commit b08495b6f6
7 changed files with 450 additions and 34 deletions
Binary file not shown.
+63 -31
View File
@@ -1,6 +1,7 @@
package main
import (
"bufio"
"context"
"crypto/md5"
"encoding/hex"
@@ -182,19 +183,22 @@ func (c *Collector) lookup(ctx context.Context, barcode string) (*Product, error
// ---- job / collector state ----
const collectedFile = "collected.txt"
type Stats struct {
Running bool `json:"running"`
Total int64 `json:"total"`
Done int64 `json:"done"`
Hits int64 `json:"hits"`
Miss int64 `json:"miss"`
Invalid int64 `json:"invalid"`
Errors int64 `json:"errors"`
Skipped int64 `json:"skipped"`
Current string `json:"current"`
OutFile string `json:"out_file"`
StartedAt string `json:"started_at"`
Message string `json:"message"`
Running bool `json:"running"`
Total int64 `json:"total"`
Done int64 `json:"done"`
Hits int64 `json:"hits"`
Miss int64 `json:"miss"`
Invalid int64 `json:"invalid"`
Errors int64 `json:"errors"`
Skipped int64 `json:"skipped"`
Current string `json:"current"`
OutFile string `json:"out_file"`
StartedAt string `json:"started_at"`
Message string `json:"message"`
HistoryCount int `json:"history_count"`
}
type Collector struct {
@@ -215,8 +219,9 @@ type Collector struct {
startedAt string
message string
seen map[string]struct{} // barcodes already in output (dedupe / resume)
recent []Product // ring of last results for UI
seen map[string]struct{} // barcodes already collected (dedupe)
collectedFp *os.File // persistent history file handle
recent []Product // ring of last results for UI
}
func NewCollector(sdogID string) *Collector {
@@ -227,6 +232,7 @@ func NewCollector(sdogID string) *Collector {
seen: map[string]struct{}{},
}
c.current.Store("")
c.loadCollected()
return c
}
@@ -238,20 +244,22 @@ func (c *Collector) snapshot() Stats {
msg := c.message
out := c.outPath
started := c.startedAt
hist := len(c.seen)
c.mu.Unlock()
return Stats{
Running: c.isRunning(),
Total: atomic.LoadInt64(&c.total),
Done: atomic.LoadInt64(&c.done),
Hits: atomic.LoadInt64(&c.hits),
Miss: atomic.LoadInt64(&c.miss),
Invalid: atomic.LoadInt64(&c.invalid),
Errors: atomic.LoadInt64(&c.errors),
Skipped: atomic.LoadInt64(&c.skipped),
Current: cur,
OutFile: out,
StartedAt: started,
Message: msg,
Running: c.isRunning(),
Total: atomic.LoadInt64(&c.total),
Done: atomic.LoadInt64(&c.done),
Hits: atomic.LoadInt64(&c.hits),
Miss: atomic.LoadInt64(&c.miss),
Invalid: atomic.LoadInt64(&c.invalid),
Errors: atomic.LoadInt64(&c.errors),
Skipped: atomic.LoadInt64(&c.skipped),
Current: cur,
OutFile: out,
StartedAt: started,
Message: msg,
HistoryCount: hist,
}
}
@@ -272,9 +280,24 @@ func (c *Collector) pushRecent(p Product) {
c.mu.Unlock()
}
// loadSeen reads an existing output file to build the dedupe set (for resume).
// loadCollected reads the persistent history file (one barcode per line).
func (c *Collector) loadCollected() {
f, err := os.Open(collectedFile)
if err != nil {
return
}
defer f.Close()
sc := bufio.NewScanner(f)
for sc.Scan() {
bc := strings.TrimSpace(sc.Text())
if bc != "" {
c.seen[bc] = struct{}{}
}
}
}
// loadSeen reads an existing output file and adds to the dedupe set (for resume).
func (c *Collector) loadSeen(path string) error {
c.seen = map[string]struct{}{}
f, err := os.Open(path)
if err != nil {
if os.IsNotExist(err) {
@@ -387,6 +410,7 @@ func (c *Collector) Start(req JobReq) error {
}
c.outFile = f
c.outPath = abs
c.collectedFp, _ = os.OpenFile(collectedFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
// reset counters
atomic.StoreInt64(&c.total, int64(len(barcodes)))
@@ -424,6 +448,11 @@ func (c *Collector) run(ctx context.Context, barcodes []string, req JobReq) {
c.outFile.Close()
c.outFile = nil
}
if c.collectedFp != nil {
c.collectedFp.Sync()
c.collectedFp.Close()
c.collectedFp = nil
}
c.current.Store("")
}()
@@ -472,14 +501,17 @@ func (c *Collector) run(ctx context.Context, barcodes []string, req JobReq) {
}
atomic.AddInt64(&c.done, 1)
c.pushRecent(*p)
writeMu.Lock()
if p.Status == "hit" || req.LogMiss {
line, _ := json.Marshal(p)
writeMu.Lock()
c.outFile.Write(line)
c.outFile.Write([]byte("\n"))
c.seen[bc] = struct{}{}
writeMu.Unlock()
}
c.seen[bc] = struct{}{}
if c.collectedFp != nil {
c.collectedFp.WriteString(bc + "\n")
}
writeMu.Unlock()
}
}
+76
View File
@@ -2,6 +2,7 @@ package main
import (
"bufio"
"bytes"
"embed"
"encoding/csv"
"encoding/json"
@@ -15,6 +16,7 @@ import (
"os"
"os/exec"
"runtime"
"strings"
"time"
)
@@ -70,6 +72,7 @@ func main() {
mux.HandleFunc("/api/stats", handleStats)
mux.HandleFunc("/api/download", handleDownload)
mux.HandleFunc("/api/export.csv", handleExportCSV)
mux.HandleFunc("/api/import", handleImport)
ln, err := net.Listen("tcp", *addr)
if err != nil {
@@ -168,6 +171,79 @@ func handleDownload(w http.ResponseWriter, r *http.Request) {
io.Copy(w, f)
}
func handleImport(w http.ResponseWriter, r *http.Request) {
if r.Method != "POST" {
http.Error(w, "method", 405)
return
}
var req struct {
APIURL string `json:"api_url"`
Username string `json:"username"`
Password string `json:"password"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeJSON(w, 400, map[string]string{"error": "请求格式错误"})
return
}
req.APIURL = strings.TrimRight(req.APIURL, "/")
if req.APIURL == "" || req.Username == "" || req.Password == "" {
writeJSON(w, 400, map[string]string{"error": "请填写完整的 API 地址、用户名和密码"})
return
}
s := collector.snapshot()
if s.OutFile == "" {
writeJSON(w, 400, map[string]string{"error": "没有采集数据可导入"})
return
}
client := &http.Client{Timeout: 60 * time.Second}
loginBody, _ := json.Marshal(map[string]string{"username": req.Username, "password": req.Password})
loginResp, err := client.Post(req.APIURL+"/api/login", "application/json", bytes.NewReader(loginBody))
if err != nil {
writeJSON(w, 502, map[string]string{"error": "无法连接 Goods 系统: " + err.Error()})
return
}
defer loginResp.Body.Close()
var loginResult struct {
Token string `json:"token"`
Error *struct {
Message string `json:"message"`
} `json:"error"`
}
json.NewDecoder(loginResp.Body).Decode(&loginResult)
if loginResp.StatusCode != 200 || loginResult.Token == "" {
msg := "登录失败"
if loginResult.Error != nil {
msg = loginResult.Error.Message
}
writeJSON(w, 401, map[string]string{"error": msg})
return
}
jsonlData, err := os.ReadFile(s.OutFile)
if err != nil {
writeJSON(w, 500, map[string]string{"error": "读取采集文件失败: " + err.Error()})
return
}
importReq, _ := http.NewRequest("POST", req.APIURL+"/api/import/bypos", bytes.NewReader(jsonlData))
importReq.Header.Set("Content-Type", "application/x-ndjson")
importReq.Header.Set("Authorization", "Bearer "+loginResult.Token)
importResp, err := client.Do(importReq)
if err != nil {
writeJSON(w, 502, map[string]string{"error": "导入请求失败: " + err.Error()})
return
}
defer importResp.Body.Close()
respBody, _ := io.ReadAll(importResp.Body)
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(importResp.StatusCode)
w.Write(respBody)
}
func handleExportCSV(w http.ResponseWriter, r *http.Request) {
s := collector.snapshot()
if s.OutFile == "" {
+62 -3
View File
@@ -37,6 +37,11 @@
.hit { color:#2da44e; } .miss { color:#999; } .invalid { color:#d1242f; } .error { color:#bf8700; }
.hint { font-size:12px; color:#888; margin-top:6px; line-height:1.5; }
.est { font-size:13px; color:#1f6feb; margin-top:6px; }
.history-badge { display:inline-block; background:#e8f0fe; color:#1f6feb; padding:4px 12px; border-radius:12px; font-size:13px; font-weight:600; margin-bottom:10px; }
.import-result { margin-top:10px; padding:10px; border-radius:6px; font-size:13px; }
.import-result.ok { background:#dcfce7; color:#166534; }
.import-result.fail { background:#fee2e2; color:#991b1b; }
input[type=password] { width:100%; padding:8px 10px; border:1px solid #cdd5e0; border-radius:6px; font-size:14px; }
</style>
</head>
<body>
@@ -97,11 +102,12 @@
</div>
</div>
<label style="margin-top:12px"><input type="checkbox" id="logmiss"/> 同时记录未命中/无效条码(默认只存命中)</label>
<div class="hint">速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(断点续采)。</div>
<div class="hint">速度越快越容易触发上游频控。建议并发 3、间隔 300ms 起步,稳定后再调。已采过的条码会自动跳过(跨次去重)。</div>
</div>
<div class="card">
<h3>③ 运行</h3>
<div class="history-badge" id="history-badge">历史已采集: 0 个条码</div>
<div style="margin-bottom:12px">
<button class="primary" id="btnStart" onclick="start()">开始采集</button>
<button class="danger" id="btnStop" onclick="stop()" disabled>停止</button>
@@ -116,13 +122,36 @@
<div class="stat"><div class="n miss" id="s-miss">0</div><div class="l">未命中</div></div>
<div class="stat"><div class="n invalid" id="s-invalid">0</div><div class="l">无效</div></div>
<div class="stat"><div class="n error" id="s-errors">0</div><div class="l">错误</div></div>
<div class="stat"><div class="n" id="s-skipped">0</div><div class="l">跳过</div></div>
<div class="stat"><div class="n" id="s-skipped">0</div><div class="l">跳过(已采)</div></div>
</div>
<div class="hint" id="msg"></div>
</div>
<div class="card">
<h3>实时结果(最近 60 条)</h3>
<h3>导入到 Goods 系统</h3>
<div class="hint" style="margin-top:0;margin-bottom:10px">采集完成后,将数据一键导入到 OpenGoods 商品档案系统。</div>
<div class="row">
<div>
<label>Goods 管理后台地址</label>
<input type="text" id="import-url" placeholder="如 http://192.168.1.100:8080/ping"/>
</div>
<div>
<label>管理员用户名</label>
<input type="text" id="import-user" placeholder="admin"/>
</div>
<div>
<label>管理员密码</label>
<input type="password" id="import-pass"/>
</div>
</div>
<div style="margin-top:12px">
<button class="primary" id="btnImport" onclick="doImport()">导入到 Goods</button>
</div>
<div id="import-result"></div>
</div>
<div class="card">
<h3>⑤ 实时结果(最近 60 条)</h3>
<div style="max-height:340px; overflow:auto">
<table>
<thead><tr><th>条码</th><th>品名</th><th>规格</th><th>单位</th><th>产地</th><th>进价</th><th>零售价</th><th>状态</th></tr></thead>
@@ -196,6 +225,7 @@ async function poll(){
document.getElementById('s-skipped').textContent = s.skipped;
let pct = s.total>0 ? Math.floor(s.done*100/s.total) : 0;
document.getElementById('prog').style.width = pct + '%';
document.getElementById('history-badge').textContent = '历史已采集: ' + (s.history_count||0) + ' 个条码';
document.getElementById('msg').textContent = (s.running? ('采集中… 当前 '+s.current) : (s.message||'空闲'));
document.getElementById('btnStart').disabled = s.running;
document.getElementById('btnStop').disabled = !s.running;
@@ -206,6 +236,35 @@ async function poll(){
}catch(e){}
}
setInterval(poll, 1000); poll();
async function doImport(){
let url = document.getElementById('import-url').value.trim();
let user = document.getElementById('import-user').value.trim();
let pass = document.getElementById('import-pass').value;
if(!url || !user || !pass){ alert('请填写完整的 Goods 系统地址、用户名和密码'); return; }
let btn = document.getElementById('btnImport');
let res = document.getElementById('import-result');
btn.disabled = true;
btn.textContent = '导入中...';
res.innerHTML = '';
try {
let r = await fetch('/api/import', {
method:'POST',
headers:{'Content-Type':'application/json'},
body:JSON.stringify({api_url:url, username:user, password:pass})
});
let j = await r.json();
if(j.error){
res.innerHTML = '<div class="import-result fail">导入失败: '+esc(j.error.message||j.error)+'</div>';
} else {
res.innerHTML = '<div class="import-result ok">导入完成: 成功 '+j.loaded+' 条, 跳过 '+j.skipped+' 条, 失败 '+j.errored+' 条</div>';
}
} catch(e){
res.innerHTML = '<div class="import-result fail">导入失败: '+esc(e.message)+'</div>';
}
btn.disabled = false;
btn.textContent = '导入到 Goods';
}
</script>
</body>
</html>