姚章浩
3e7fcce0b1
build / build (push) Successful in 1m11s
新增迁移50与每5分钟/手动持久同步，API只读本地；同步失败保留、权限阻断、租约与质量失效原子激活。 保留全量稳定分页/固定列，已存质量批量读取，元数据不反复解压长答案；详情安全Markdown与原文。 已运行整仓 make verify、源与副本隔离PG -race、真实只读源及隔离浏览器。 实际同步2525条；固定窗口640条本地中位1.46秒，对照直接源33.10秒。 停worker、缺源凭据、重启API后仍可查询；未更新8088，未修改生产原问答。
285 lines
8.8 KiB
Go
285 lines
8.8 KiB
Go
package httpapi
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"io"
|
||
"knowledge-governance/internal/dobotaiops"
|
||
"knowledge-governance/internal/qaopsquality"
|
||
"net/http"
|
||
"strconv"
|
||
"time"
|
||
)
|
||
|
||
type QAQualityStore interface {
|
||
ObserveKeys(context.Context, []dobotaiops.QualityKey) (map[string]qaopsquality.View, error)
|
||
Observe(context.Context, []dobotaiops.Record) (map[string]qaopsquality.View, error)
|
||
Save(context.Context, dobotaiops.Record, qaopsquality.SaveRequest, string) (qaopsquality.Receipt, error)
|
||
Columns(context.Context, []dobotaiops.QualityKey) ([]string, error)
|
||
History(context.Context, string, int64, int64) (qaopsquality.History, error)
|
||
Snapshot(context.Context, string, int64) (qaopsquality.Snapshot, error)
|
||
}
|
||
|
||
type storedQAQualityReader interface {
|
||
ReadStoredQuality(context.Context, []dobotaiops.QualityKey) (map[string]qaopsquality.View, error)
|
||
}
|
||
|
||
func qaQualityViews(ctx context.Context, c Config, records []dobotaiops.Record) (map[string]qaopsquality.View, error) {
|
||
if reader, ok := c.QAOps.(storedQAQualityReader); ok {
|
||
keys := make([]dobotaiops.QualityKey, 0, len(records))
|
||
for _, r := range records {
|
||
keys = append(keys, dobotaiops.QualityKey{ID: r.ID, InputHash: r.QualityInputHash})
|
||
}
|
||
return reader.ReadStoredQuality(ctx, keys)
|
||
}
|
||
return c.QAQuality.Observe(ctx, records)
|
||
}
|
||
|
||
type qaQualityReader interface {
|
||
ReadQualityScope(context.Context, dobotaiops.RecordQuery) (dobotaiops.QualityScope, error)
|
||
}
|
||
|
||
func writeQualityError(w http.ResponseWriter, err error) {
|
||
status, code, message := 503, "quality_unavailable", "质量留档/复核存储不可用，未将失败伪装为空值或零统计"
|
||
switch {
|
||
case errors.Is(err, qaopsquality.ErrInvalid) || errors.Is(err, dobotaiops.ErrInvalidQuery):
|
||
status, code, message = 400, "invalid_quality_request", "质量操作参数无效：请核对结论、理由、版本和筛选范围（最多5000条/32MiB）"
|
||
case errors.Is(err, qaopsquality.ErrConflict):
|
||
status, code, message = 409, "quality_conflict", "原始问答或人工复核版本已变化，请重新读取问答后核对；未覆盖现有结论"
|
||
case errors.Is(err, qaopsquality.ErrSnapshotMissing):
|
||
status, code, message = 409, "quality_input_metadata_only", "此版本仅留存指纹与复检元数据，未采集原文；未编造空问题或空答案"
|
||
case errors.Is(err, qaopsquality.ErrNotFound):
|
||
status, code, message = 404, "quality_input_not_found", "未找到当前授权范围的输入版本"
|
||
}
|
||
writeJSON(w, status, map[string]string{"error": message, "code": code})
|
||
}
|
||
func enrichQAQualityRecord(ctx context.Context, c Config, item *dobotaiops.Record) error {
|
||
if c.QAQuality == nil {
|
||
return nil
|
||
}
|
||
views, err := qaQualityViews(ctx, c, []dobotaiops.Record{*item})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
v, ok := views[item.ID]
|
||
if !ok {
|
||
return qaopsquality.ErrUnavailable
|
||
}
|
||
item.Quality, err = json.Marshal(v)
|
||
return err
|
||
}
|
||
func enrichQAQualityPage(ctx context.Context, c Config, page *dobotaiops.RecordPage) error {
|
||
if c.QAQuality == nil {
|
||
return nil
|
||
}
|
||
views, err := qaQualityViews(ctx, c, page.Items)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for i := range page.Items {
|
||
v, ok := views[page.Items[i].ID]
|
||
if !ok {
|
||
return qaopsquality.ErrUnavailable
|
||
}
|
||
page.Items[i].Quality, err = json.Marshal(v)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
columns, err := c.QAQuality.Columns(ctx, page.ScopeQuality)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
dobotaiops.AddRecordColumns(page, columns...)
|
||
return nil
|
||
}
|
||
func walkQualityJSON(d *json.Decoder) error {
|
||
token, err := d.Token()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if token == nil {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
if delimiter, ok := token.(json.Delim); ok {
|
||
if delimiter != '{' && delimiter != '[' {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
keys := map[string]bool{}
|
||
for d.More() {
|
||
if delimiter == '{' {
|
||
key, e := d.Token()
|
||
name, ok := key.(string)
|
||
if e != nil || !ok || keys[name] {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
keys[name] = true
|
||
}
|
||
if walkQualityJSON(d) != nil {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
}
|
||
_, err = d.Token()
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
func decodeQualityJSON(w http.ResponseWriter, r *http.Request, target any, required ...string) error {
|
||
if r.URL.RawQuery != "" {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 64<<10))
|
||
if err != nil || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
var top map[string]json.RawMessage
|
||
if json.Unmarshal(raw, &top) != nil || top == nil {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
for _, key := range required {
|
||
if _, ok := top[key]; !ok {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
}
|
||
check := json.NewDecoder(bytes.NewReader(raw))
|
||
if walkQualityJSON(check) != nil {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
var extra any
|
||
if check.Decode(&extra) != io.EOF {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
d := json.NewDecoder(bytes.NewReader(raw))
|
||
d.DisallowUnknownFields()
|
||
if d.Decode(target) != nil || d.Decode(&extra) != io.EOF {
|
||
return qaopsquality.ErrInvalid
|
||
}
|
||
return nil
|
||
}
|
||
func qaQualityReview(w http.ResponseWriter, r *http.Request, c Config) {
|
||
if !authorizeQAOps(w, r) || !authorizeQAIntent(w, r, "qa-quality") {
|
||
return
|
||
}
|
||
var request qaopsquality.SaveRequest
|
||
if decodeQualityJSON(w, r, &request, "input_hash", "input_epoch", "base_revision", "nonce", "fields", "reason") != nil || qaopsquality.Validate(request) != nil {
|
||
writeQualityError(w, qaopsquality.ErrInvalid)
|
||
return
|
||
}
|
||
if c.QAQuality == nil || c.QAOps == nil || c.QAOpsTenantKey == "" {
|
||
writeQualityError(w, qaopsquality.ErrUnavailable)
|
||
return
|
||
}
|
||
record, err := c.QAOps.GetRecord(r.Context(), c.QAOpsTenantKey, r.PathValue("id"))
|
||
if err != nil {
|
||
qaOpsError(w, err)
|
||
return
|
||
}
|
||
identity, _ := sessionFrom(r.Context())
|
||
receipt, err := c.QAQuality.Save(r.Context(), record, request, identity.Username)
|
||
if err != nil {
|
||
writeQualityError(w, err)
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"data": receipt})
|
||
}
|
||
func qaQualityQuery(w http.ResponseWriter, r *http.Request, c Config) {
|
||
if !authorizeQAOps(w, r) || !authorizeQAIntent(w, r, "qa-quality") {
|
||
return
|
||
}
|
||
var request struct {
|
||
Start string `json:"start"`
|
||
End string `json:"end"`
|
||
Keyword string `json:"keyword"`
|
||
}
|
||
if decodeQualityJSON(w, r, &request) != nil {
|
||
writeQualityError(w, qaopsquality.ErrInvalid)
|
||
return
|
||
}
|
||
window, err := dobotaiops.NewWindow(c.QAOpsTenantKey, request.Start, request.End, 90*24*time.Hour, c.Now())
|
||
if err != nil {
|
||
writeQualityError(w, err)
|
||
return
|
||
}
|
||
reader, ok := c.QAOps.(qaQualityReader)
|
||
if !ok || c.QAQuality == nil {
|
||
writeQualityError(w, qaopsquality.ErrUnavailable)
|
||
return
|
||
}
|
||
scope, err := reader.ReadQualityScope(r.Context(), dobotaiops.RecordQuery{Window: window, Keyword: request.Keyword})
|
||
if err != nil {
|
||
qaOpsError(w, err)
|
||
return
|
||
}
|
||
var views map[string]qaopsquality.View
|
||
if stored, ok := c.QAOps.(storedQAQualityReader); ok {
|
||
views, err = stored.ReadStoredQuality(r.Context(), scope.Keys)
|
||
} else {
|
||
views, err = c.QAQuality.ObserveKeys(r.Context(), scope.Keys)
|
||
}
|
||
if err != nil || len(views) != len(scope.Keys) {
|
||
if err == nil {
|
||
err = qaopsquality.ErrUnavailable
|
||
}
|
||
writeQualityError(w, err)
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"data": map[string]any{"summary": qaopsquality.Summarize(views), "start": window.Start, "end": window.End, "keyword": request.Keyword, "as_of": scope.AsOf, "ai_state": "not_configured", "scope": "whole_filter"}})
|
||
}
|
||
func qaQualityHistory(w http.ResponseWriter, r *http.Request, c Config) {
|
||
if !authorizeQAOps(w, r) {
|
||
return
|
||
}
|
||
if c.QAQuality == nil {
|
||
writeQualityError(w, qaopsquality.ErrUnavailable)
|
||
return
|
||
}
|
||
values := r.URL.Query()
|
||
for key, v := range values {
|
||
if (key != "review_before" && key != "check_before") || len(v) != 1 {
|
||
writeQualityError(w, qaopsquality.ErrInvalid)
|
||
return
|
||
}
|
||
}
|
||
cursors := map[string]int64{}
|
||
for _, key := range []string{"review_before", "check_before"} {
|
||
raw := values.Get(key)
|
||
if raw == "" {
|
||
continue
|
||
}
|
||
n, err := strconv.ParseInt(raw, 10, 64)
|
||
if err != nil || n < 1 {
|
||
writeQualityError(w, qaopsquality.ErrInvalid)
|
||
return
|
||
}
|
||
cursors[key] = n
|
||
}
|
||
value, err := c.QAQuality.History(r.Context(), r.PathValue("id"), cursors["review_before"], cursors["check_before"])
|
||
if err != nil {
|
||
writeQualityError(w, err)
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"data": value})
|
||
}
|
||
func qaQualitySnapshot(w http.ResponseWriter, r *http.Request, c Config) {
|
||
if !authorizeQAOps(w, r) {
|
||
return
|
||
}
|
||
epoch, err := strconv.ParseInt(r.PathValue("epoch"), 10, 64)
|
||
if err != nil || epoch < 1 || r.URL.RawQuery != "" {
|
||
writeQualityError(w, qaopsquality.ErrInvalid)
|
||
return
|
||
}
|
||
if c.QAQuality == nil {
|
||
writeQualityError(w, qaopsquality.ErrUnavailable)
|
||
return
|
||
}
|
||
value, err := c.QAQuality.Snapshot(r.Context(), r.PathValue("id"), epoch)
|
||
if err != nil {
|
||
writeQualityError(w, err)
|
||
return
|
||
}
|
||
writeJSON(w, 200, map[string]any{"data": value})
|
||
}
|