From 461fecd9922e280a6946069fee44e111f39bf576 Mon Sep 17 00:00:00 2001 From: walker1211 <13750528578@163.com> Date: Tue, 11 Aug 2026 22:10:06 +0800 Subject: [PATCH] =?UTF-8?q?perf(pipeline):=20=E9=99=8D=E4=BD=8E=E6=8A=93?= =?UTF-8?q?=E5=8F=96=E5=BC=80=E9=94=80=E5=B9=B6=E7=BB=9F=E4=B8=80=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E6=A0=87=E8=AF=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.zh-CN.md | 4 + cmds/news-briefing/execute.go | 1 + cmds/news-briefing/scheduled_state.go | 16 ++- configs/config.example.yaml | 4 + internal/config/config.go | 33 ++++- internal/config/config_test.go | 52 +++++++ internal/fetcher/fetch.go | 19 ++- internal/fetcher/rss.go | 15 +- internal/fetcher/rss_cache.go | 192 ++++++++++++++++++++++++++ internal/fetcher/rss_cache_test.go | 50 +++++++ internal/model/source_stats.go | 27 ++-- internal/watch/announcement_site.go | 20 +-- internal/watch/run.go | 28 ++-- internal/watch/run_test.go | 98 +++++++++++-- internal/watch/site_article_state.go | 53 +++++-- 15 files changed, 543 insertions(+), 69 deletions(-) create mode 100644 internal/fetcher/rss_cache.go create mode 100644 internal/fetcher/rss_cache_test.go diff --git a/README.zh-CN.md b/README.zh-CN.md index 584c65a..2109762 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -324,6 +324,10 @@ schedule_delay: 10m 服务实际观察到 08:00 / 18:00 触发时，会把该窗口登记到唯一的长期状态文件 `output/state/briefing-scheduler.json`。只有处于 `pending`、`waiting_x` 或 `running` 的已登记窗口才有 watcher；默认每 1 分钟检查一次，进入 `done` / `failed` 后立即停止。若同窗口的 X 状态仍为 `running` 且 heartbeat 新鲜，窗口切到 `waiting_x`；X 进入终态或 heartbeat 超过 3 分钟未更新时，watcher 接管执行。简报自身也每分钟更新 heartbeat，超过 3 分钟可由重启后的 watcher 接管。cron、X 回调和 watcher 通过短期文件锁原子竞争同一窗口的 lease，所以同一窗口只会有一个有效执行者；旧 lease 不能覆盖接管后的状态。邮件成功时间也保存在同一记录中，接管时不会重复发送已经确认成功的邮件。短期 `.lock` 和原子写临时文件只在更新状态时存在，不是长期 marker。 +Watch 默认走“索引快检 + 正文深检”：索引新增或变化会立即读取正文；未变化文章按 `watch.deep_verify_interval` 到期后，以 `watch.deep_verify_batch_size` 为上限按最旧检查时间轮转。这样仍能发现 URL、标题和摘要均未变化时的正文静默更新，同时避免每个简报窗口下载全部历史正文。 + +RSS 源会在 `/state/rss-cache` 保存压缩响应和 ETag/Last-Modified 元数据。服务端返回 `304 Not Modified` 时复用已缓存 Feed；每份 `.source-stats.json` 同时记录各来源的抓取耗时、响应字节数与缓存状态，便于识别大 Feed 和低有效率来源。 + 注意：`serve` 启动时只恢复上述状态文件里尚未结束的窗口，不会推算或补跑服务完全错过的历史触发点。例如 07:50 停服、08:01 启动时不会自动创建 08:00 窗口，按需使用 `regen` 手动补跑。 建议：调整 cron / `schedule` 后，如怀疑有断层，优先使用项目自带的 `regen --from --to` 手动补窗，例如： diff --git a/cmds/news-briefing/execute.go b/cmds/news-briefing/execute.go index 20d6bb5..f1524f6 100644 --- a/cmds/news-briefing/execute.go +++ b/cmds/news-briefing/execute.go @@ -92,6 +92,7 @@ type emailDeps struct { func newApp(cfg *config.Config) *app { httpClient := fetcher.NewHTTPClient(cfg.Proxy, cfg.Fetch.Timeout) fetchClient := fetcher.NewClient(httpClient) + fetchClient.SetRSSCacheDir(filepath.Join(cfg.Output.Dir, "state", "rss-cache")) watchRunner := watch.NewRunner(httpClient) aiRunner := summarizer.NewRunnerWithRetryDelays(cfg.AI.Command, cfg.AI.Args, cfg.AI.ShouldAppendSystemPrompt(), cfg.Proxy.HTTP, cfg.Proxy.Socks5, cfg.AI.Retry.Delays) aiRunner.SetModels(cfg.AI.Models.Default, cfg.AI.Models.Translation) diff --git a/cmds/news-briefing/scheduled_state.go b/cmds/news-briefing/scheduled_state.go index 3cab348..f18b9ca 100644 --- a/cmds/news-briefing/scheduled_state.go +++ b/cmds/news-briefing/scheduled_state.go @@ -43,6 +43,7 @@ type scheduledStateFile struct { } type scheduledWindowState struct { + RunID string `json:"runId,omitempty"` Expr string `json:"expr"` Period string `json:"period"` From time.Time `json:"from"` @@ -162,6 +163,10 @@ func scheduledRunWindowKey(window scheduler.Window) string { return window.From.UTC().Format("20060102T150405Z") + "_" + window.To.UTC().Format("20060102T150405Z") } +func scheduledRunID(window scheduler.Window) string { + return scheduledRunWindowKey(window) +} + func scheduledWindowFromState(record scheduledWindowState) (scheduler.Window, error) { if record.Period == "" || record.From.IsZero() || !record.To.After(record.From) { return scheduler.Window{}, fmt.Errorf("invalid scheduled window state") @@ -306,6 +311,7 @@ func (app *app) registerScheduledWindow(window scheduler.Window, dueAt time.Time return false, nil } state.Windows[key] = scheduledWindowState{ + RunID: scheduledRunID(window), Expr: window.Expr, Period: window.Period, From: window.From, @@ -338,7 +344,10 @@ func (app *app) markScheduledRunWaitingX(window scheduler.Window, trigger string return false, nil } if !ok { - record = scheduledWindowState{Expr: window.Expr, Period: window.Period, From: window.From, To: window.To, DueAt: window.To} + record = scheduledWindowState{RunID: scheduledRunID(window), Expr: window.Expr, Period: window.Period, From: window.From, To: window.To, DueAt: window.To} + } + if record.RunID == "" { + record.RunID = scheduledRunID(window) } record.Status = scheduledStatusWaitingX record.Trigger = trigger @@ -378,7 +387,10 @@ func (app *app) acquireScheduledRunWindow(window scheduler.Window, trigger strin return false, nil } if !ok { - record = scheduledWindowState{Expr: window.Expr, Period: window.Period, From: window.From, To: window.To, DueAt: window.To} + record = scheduledWindowState{RunID: scheduledRunID(window), Expr: window.Expr, Period: window.Period, From: window.From, To: window.To, DueAt: window.To} + } + if record.RunID == "" { + record.RunID = scheduledRunID(window) } emailAlreadySent = !record.EmailSentAt.IsZero() record.Status = scheduledStatusRunning diff --git a/configs/config.example.yaml b/configs/config.example.yaml index f54ebbf..071e45d 100644 --- a/configs/config.example.yaml +++ b/configs/config.example.yaml @@ -370,6 +370,10 @@ fetch: watch: article_concurrency: 8 + # Normal runs only fetch changed/new article bodies. A bounded oldest-first + # audit still detects silent body changes without downloading the full corpus. + deep_verify_interval: 24h + deep_verify_batch_size: 48 sites: - name: Anthropic Claude Support type: anthropic_support diff --git a/internal/config/config.go b/internal/config/config.go index decf758..6214f3c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -61,6 +61,8 @@ const ( DefaultWatchBrowseboxProxyPort = 17997 DefaultWatchBrowseboxControllerPort = 17998 DefaultWatchArticleConcurrency = 8 + DefaultWatchDeepVerifyInterval = 24 * time.Hour + DefaultWatchDeepVerifyBatchSize = 48 DefaultXRefreshWaitTimeout = 10 * time.Minute DefaultXRefreshWaitInterval = 5 * time.Second DefaultXRefreshReconcileInterval = time.Minute @@ -137,10 +139,14 @@ type FetchConfig struct { } type WatchConfig struct { - Sites []WatchSite `yaml:"sites"` - ArticleConcurrencyRaw *int `yaml:"article_concurrency"` - ArticleConcurrency int `yaml:"-"` - ProxyProvider WatchProxyProvider `yaml:"proxy_provider"` + Sites []WatchSite `yaml:"sites"` + ArticleConcurrencyRaw *int `yaml:"article_concurrency"` + ArticleConcurrency int `yaml:"-"` + DeepVerifyIntervalRaw string `yaml:"deep_verify_interval"` + DeepVerifyInterval time.Duration `yaml:"-"` + DeepVerifyBatchSizeRaw *int `yaml:"deep_verify_batch_size"` + DeepVerifyBatchSize int `yaml:"-"` + ProxyProvider WatchProxyProvider `yaml:"proxy_provider"` } type WatchProxyProvider struct { @@ -359,6 +365,19 @@ func applyWatchDefaults(watch *WatchConfig) error { } else { watch.ArticleConcurrency = *watch.ArticleConcurrencyRaw } + if strings.TrimSpace(watch.DeepVerifyIntervalRaw) == "" { + watch.DeepVerifyIntervalRaw = DefaultWatchDeepVerifyInterval.String() + } + deepVerifyInterval, err := time.ParseDuration(strings.TrimSpace(watch.DeepVerifyIntervalRaw)) + if err != nil { + return fmt.Errorf("parse watch.deep_verify_interval: %w", err) + } + watch.DeepVerifyInterval = deepVerifyInterval + if watch.DeepVerifyBatchSizeRaw == nil { + watch.DeepVerifyBatchSize = DefaultWatchDeepVerifyBatchSize + } else { + watch.DeepVerifyBatchSize = *watch.DeepVerifyBatchSizeRaw + } provider := &watch.ProxyProvider if !provider.Enabled { return nil @@ -639,6 +658,12 @@ func (cfg *Config) Validate() error { if cfg.Watch.ArticleConcurrency < 1 { return fmt.Errorf("validate watch.article_concurrency: must be at least 1") } + if cfg.Watch.DeepVerifyInterval <= 0 { + return fmt.Errorf("validate watch.deep_verify_interval: must be greater than 0") + } + if cfg.Watch.DeepVerifyBatchSize < 1 { + return fmt.Errorf("validate watch.deep_verify_batch_size: must be at least 1") + } for i, site := range cfg.Watch.Sites { if err := validateWatchSite(i, site); err != nil { return err diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 9b3ab3f..316670f 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1508,6 +1508,58 @@ ai: {} } } +func TestLoadAppliesWatchDeepVerificationDefaults(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + content := `sources: [] +keywords: [] +fetch: {} +watch: {} +email: {} +schedule: [] +output: {} +proxy: {} +ai: {} +` + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatalf("write config: %v", err) + } + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if cfg.Watch.DeepVerifyInterval != 24*time.Hour || cfg.Watch.DeepVerifyBatchSize != 48 { + t.Fatalf("watch deep verification = %s/%d", cfg.Watch.DeepVerifyInterval, cfg.Watch.DeepVerifyBatchSize) + } +} + +func TestLoadAppliesConfiguredWatchDeepVerification(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.yaml") + content := `sources: [] +keywords: [] +fetch: {} +watch: + deep_verify_interval: 12h + deep_verify_batch_size: 16 +email: {} +schedule: [] +output: {} +proxy: {} +ai: {} +` + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatalf("write config: %v", err) + } + cfg, err := Load(path) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if cfg.Watch.DeepVerifyInterval != 12*time.Hour || cfg.Watch.DeepVerifyBatchSize != 16 { + t.Fatalf("watch deep verification = %s/%d", cfg.Watch.DeepVerifyInterval, cfg.Watch.DeepVerifyBatchSize) + } +} + func TestLoadRejectsInvalidWatchArticleConcurrency(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "config.yaml") diff --git a/internal/fetcher/fetch.go b/internal/fetcher/fetch.go index c6e04b1..4325b3d 100644 --- a/internal/fetcher/fetch.go +++ b/internal/fetcher/fetch.go @@ -103,6 +103,9 @@ type sourceFetchResult struct { Candidates []fetchedCandidate FetchedCount int RedditRateLimitWait time.Duration + FetchDuration time.Duration + ResponseBytes int64 + CacheStatus string } type sourceFetchFunc func(context.Context, config.Source, []string, time.Time) (sourceFetchResult, error) @@ -121,6 +124,7 @@ type curlFetchFunc func(context.Context, string) ([]byte, error) type Client struct { httpClient *http.Client fetchCurl curlFetchFunc + rssCache *rssFeedCache } func NewClient(httpClient *http.Client) *Client { @@ -130,6 +134,13 @@ func NewClient(httpClient *http.Client) *Client { return &Client{httpClient: httpClient, fetchCurl: fetchFeedWithCurlContext} } +func (c *Client) SetRSSCacheDir(dir string) { + if c == nil || strings.TrimSpace(dir) == "" { + return + } + c.rssCache = newRSSFeedCache(dir) +} + func fetchRSSSource(ctx context.Context, src config.Source, keywords []string, since time.Time) (sourceFetchResult, error) { return FetchRSSContext(ctx, src, keywords, since) } @@ -367,8 +378,14 @@ func (acc *sourceStatsAccumulator) statForArticle(article model.Article, fallbac } func (acc *sourceStatsAccumulator) countFetched(result sourceFetchResult) { + entry := acc.statForArticle(model.Article{}, result.Source) + entry.FetchDurationMS += result.FetchDuration.Milliseconds() + entry.ResponseBytes += result.ResponseBytes + if result.CacheStatus != "" { + entry.CacheStatus = result.CacheStatus + } if result.FetchedCount > 0 { - acc.statForArticle(model.Article{}, result.Source).Fetched += result.FetchedCount + entry.Fetched += result.FetchedCount return } for _, candidate := range result.Candidates { diff --git a/internal/fetcher/rss.go b/internal/fetcher/rss.go index 37fad37..4aea3c5 100644 --- a/internal/fetcher/rss.go +++ b/internal/fetcher/rss.go @@ -69,7 +69,8 @@ func (c *Client) fetchRSSContextWithOpenGraphOptions(ctx context.Context, source fp := gofeed.NewParser() fp.Client = c.httpClient - feed, headers, err := c.fetchRSSFeed(ctx, fetchSource, fp) + fetchStarted := time.Now() + feed, headers, responseBytes, cacheStatus, err := c.fetchRSSFeed(ctx, fetchSource, source.URL, fp) if err != nil { if !shouldFallbackToCurl(source, err) { return sourceFetchResult{}, err @@ -87,9 +88,11 @@ func (c *Client) fetchRSSContextWithOpenGraphOptions(ctx context.Context, source return sourceFetchResult{}, err } headers = nil + responseBytes = int64(len(body)) + cacheStatus = "curl" } - result := sourceFetchResult{Source: source, FetchedCount: len(feed.Items)} + result := sourceFetchResult{Source: source, FetchedCount: len(feed.Items), FetchDuration: time.Since(fetchStarted), ResponseBytes: responseBytes, CacheStatus: cacheStatus} isRedditRSS := isRedditURL(source.URL) if isRedditRSS { result.RedditRateLimitWait = redditRateLimitWaitFromHeader(headers) @@ -185,12 +188,12 @@ func authenticatedRSSURL(source config.Source) (string, error) { return parsedURL.String(), nil } -func (c *Client) fetchRSSFeed(ctx context.Context, source config.Source, fp *gofeed.Parser) (*gofeed.Feed, http.Header, error) { +func (c *Client) fetchRSSFeed(ctx context.Context, source config.Source, cacheKeyURL string, fp *gofeed.Parser) (*gofeed.Feed, http.Header, int64, string, error) { if !isRedditURL(source.URL) { - feed, err := fp.ParseURLWithContext(source.URL, ctx) - return feed, nil, err + return c.fetchRSSFeedHTTP(ctx, source.URL, cacheKeyURL, fp) } - return c.fetchRSSFeedWithHeaders(ctx, source.URL, fp) + feed, headers, err := c.fetchRSSFeedWithHeaders(ctx, source.URL, fp) + return feed, headers, 0, "network", err } func (c *Client) fetchRSSFeedWithHeaders(ctx context.Context, feedURL string, fp *gofeed.Parser) (*gofeed.Feed, http.Header, error) { diff --git a/internal/fetcher/rss_cache.go b/internal/fetcher/rss_cache.go new file mode 100644 index 0000000..d65ec10 --- /dev/null +++ b/internal/fetcher/rss_cache.go @@ -0,0 +1,192 @@ +package fetcher + +import ( + "bytes" + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/mmcdole/gofeed" + "github.com/walker1211/news-briefing/internal/statefile" +) + +const maxRSSFeedBytes = 32 * 1024 * 1024 + +type rssFeedCache struct { + dir string + mu sync.Mutex + locks map[string]*sync.Mutex +} + +type rssFeedCacheMetadata struct { + ETag string `json:"etag,omitempty"` + LastModified string `json:"last_modified,omitempty"` +} + +func newRSSFeedCache(dir string) *rssFeedCache { + return &rssFeedCache{dir: dir, locks: make(map[string]*sync.Mutex)} +} + +func (c *Client) fetchRSSFeedHTTP(ctx context.Context, feedURL, cacheKeyURL string, parser *gofeed.Parser) (*gofeed.Feed, http.Header, int64, string, error) { + if c.rssCache == nil { + return c.fetchRSSFeedNetwork(ctx, feedURL, parser, nil, "network") + } + return c.rssCache.fetch(ctx, c.httpClient, feedURL, cacheKeyURL, parser) +} + +func (c *Client) fetchRSSFeedNetwork(ctx context.Context, feedURL string, parser *gofeed.Parser, metadata *rssFeedCacheMetadata, status string) (*gofeed.Feed, http.Header, int64, string, error) { + req, err := newRSSFeedRequest(ctx, feedURL, metadata) + if err != nil { + return nil, nil, 0, status, err + } + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, nil, 0, status, err + } + defer resp.Body.Close() + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, resp.Header, 0, status, fmt.Errorf("http error: %d %s", resp.StatusCode, resp.Status) + } + body, err := readRSSFeedBody(resp.Body) + if err != nil { + return nil, resp.Header, 0, status, err + } + feed, err := parser.Parse(bytes.NewReader(body)) + return feed, resp.Header, int64(len(body)), status, err +} + +func (cache *rssFeedCache) fetch(ctx context.Context, client *http.Client, feedURL, cacheKeyURL string, parser *gofeed.Parser) (*gofeed.Feed, http.Header, int64, string, error) { + key := rssCacheKey(cacheKeyURL) + keyLock := cache.lockFor(key) + keyLock.Lock() + defer keyLock.Unlock() + metadata, cachedBody, _ := cache.load(key) + req, err := newRSSFeedRequest(ctx, feedURL, metadata) + if err != nil { + return nil, nil, 0, "network", err + } + resp, err := client.Do(req) + if err != nil { + return nil, nil, 0, "network", err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotModified && len(cachedBody) > 0 { + feed, parseErr := parser.Parse(bytes.NewReader(cachedBody)) + return feed, resp.Header, 0, "not_modified", parseErr + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, resp.Header, 0, "network", fmt.Errorf("http error: %d %s", resp.StatusCode, resp.Status) + } + body, err := readRSSFeedBody(resp.Body) + if err != nil { + return nil, resp.Header, 0, "network", err + } + feed, err := parser.Parse(bytes.NewReader(body)) + if err != nil { + return nil, resp.Header, int64(len(body)), "network", err + } + newMetadata := rssFeedCacheMetadata{ETag: resp.Header.Get("ETag"), LastModified: resp.Header.Get("Last-Modified")} + if saveErr := cache.save(key, newMetadata, body); saveErr != nil { + return feed, resp.Header, int64(len(body)), "cache_write_failed", nil + } + return feed, resp.Header, int64(len(body)), "updated", nil +} + +func (cache *rssFeedCache) lockFor(key string) *sync.Mutex { + cache.mu.Lock() + defer cache.mu.Unlock() + if existing := cache.locks[key]; existing != nil { + return existing + } + lock := &sync.Mutex{} + cache.locks[key] = lock + return lock +} + +func newRSSFeedRequest(ctx context.Context, feedURL string, metadata *rssFeedCacheMetadata) (*http.Request, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, feedURL, nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/rss+xml, application/atom+xml, application/xml;q=0.9, */*;q=0.8") + req.Header.Set("User-Agent", userAgent) + if metadata != nil { + if metadata.ETag != "" { + req.Header.Set("If-None-Match", metadata.ETag) + } + if metadata.LastModified != "" { + req.Header.Set("If-Modified-Since", metadata.LastModified) + } + } + return req, nil +} + +func readRSSFeedBody(reader io.Reader) ([]byte, error) { + body, err := io.ReadAll(io.LimitReader(reader, maxRSSFeedBytes+1)) + if err != nil { + return nil, err + } + if len(body) > maxRSSFeedBytes { + return nil, fmt.Errorf("RSS feed exceeds %d bytes", maxRSSFeedBytes) + } + return body, nil +} + +func rssCacheKey(rawURL string) string { + sum := sha256.Sum256([]byte(strings.TrimSpace(rawURL))) + return hex.EncodeToString(sum[:]) +} + +func (cache *rssFeedCache) load(key string) (*rssFeedCacheMetadata, []byte, error) { + metadataBytes, err := os.ReadFile(filepath.Join(cache.dir, key+".json")) + if err != nil { + return nil, nil, err + } + var metadata rssFeedCacheMetadata + if err := json.Unmarshal(metadataBytes, &metadata); err != nil { + return nil, nil, err + } + file, err := os.Open(filepath.Join(cache.dir, key+".xml.gz")) + if err != nil { + return nil, nil, err + } + defer file.Close() + reader, err := gzip.NewReader(file) + if err != nil { + return nil, nil, err + } + defer reader.Close() + body, err := readRSSFeedBody(reader) + return &metadata, body, err +} + +func (cache *rssFeedCache) save(key string, metadata rssFeedCacheMetadata, body []byte) error { + if err := os.MkdirAll(cache.dir, 0o755); err != nil { + return err + } + var compressed bytes.Buffer + writer := gzip.NewWriter(&compressed) + if _, err := writer.Write(body); err != nil { + return err + } + if err := writer.Close(); err != nil { + return err + } + if err := statefile.WriteAtomic(filepath.Join(cache.dir, key+".xml.gz"), compressed.Bytes(), 0o644); err != nil { + return err + } + metadataBytes, err := json.MarshalIndent(metadata, "", " ") + if err != nil { + return err + } + return statefile.WriteAtomic(filepath.Join(cache.dir, key+".json"), append(metadataBytes, '\n'), 0o644) +} diff --git a/internal/fetcher/rss_cache_test.go b/internal/fetcher/rss_cache_test.go new file mode 100644 index 0000000..0c2fd60 --- /dev/null +++ b/internal/fetcher/rss_cache_test.go @@ -0,0 +1,50 @@ +package fetcher + +import ( + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/walker1211/news-briefing/internal/config" +) + +func TestRSSCacheUsesConditionalRequestAndCachedBody(t *testing.T) { + var requests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + if r.Header.Get("If-None-Match") == `"feed-v1"` { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("Content-Type", "application/rss+xml") + w.Header().Set("ETag", `"feed-v1"`) + _, _ = w.Write([]byte(`Feedhttps://example.comtestCached itemhttps://example.com/itemTue, 11 Aug 2026 00:00:00 GMTsummary]]>`)) + })) + defer server.Close() + + client := NewClient(server.Client()) + client.SetRSSCacheDir(t.TempDir()) + source := config.Source{Name: "cached", Type: config.SourceTypeRSS, URL: server.URL, Category: "AI"} + first, err := client.FetchRSS(source, nil, time.Time{}) + if err != nil { + t.Fatalf("first FetchRSS() error = %v", err) + } + second, err := client.FetchRSS(source, nil, time.Time{}) + if err != nil { + t.Fatalf("second FetchRSS() error = %v", err) + } + if first.CacheStatus != "updated" || first.ResponseBytes == 0 { + t.Fatalf("first metrics = cache:%q bytes:%d", first.CacheStatus, first.ResponseBytes) + } + if second.CacheStatus != "not_modified" || second.ResponseBytes != 0 { + t.Fatalf("second metrics = cache:%q bytes:%d", second.CacheStatus, second.ResponseBytes) + } + if len(second.Candidates) != 1 || second.Candidates[0].Article.Title != "Cached item" { + t.Fatalf("second candidates = %#v", second.Candidates) + } + if requests.Load() != 2 { + t.Fatalf("requests = %d, want 2", requests.Load()) + } +} diff --git a/internal/model/source_stats.go b/internal/model/source_stats.go index 24bf7c1..69e63be 100644 --- a/internal/model/source_stats.go +++ b/internal/model/source_stats.go @@ -20,16 +20,18 @@ type SourceStatsWindow struct { } type SourceStatsTotals struct { - Fetched int `json:"fetched"` - InWindow int `json:"in_window"` - KeywordMatched int `json:"keyword_matched"` - Filtered int `json:"filtered"` - FilteredKeywordMiss int `json:"filtered_keyword_miss"` - FilteredExcluded int `json:"filtered_excluded"` - FilteredSourceLimit int `json:"filtered_source_limit"` - AcceptedBeforeDedup int `json:"accepted_before_dedup"` - AcceptedAfterDedup int `json:"accepted_after_dedup"` - EnteredAI int `json:"entered_ai"` + Fetched int `json:"fetched"` + InWindow int `json:"in_window"` + KeywordMatched int `json:"keyword_matched"` + Filtered int `json:"filtered"` + FilteredKeywordMiss int `json:"filtered_keyword_miss"` + FilteredExcluded int `json:"filtered_excluded"` + FilteredSourceLimit int `json:"filtered_source_limit"` + AcceptedBeforeDedup int `json:"accepted_before_dedup"` + AcceptedAfterDedup int `json:"accepted_after_dedup"` + EnteredAI int `json:"entered_ai"` + FetchDurationMS int64 `json:"fetch_duration_ms"` + ResponseBytes int64 `json:"response_bytes"` } type SourceStatsEntry struct { @@ -46,6 +48,9 @@ type SourceStatsEntry struct { AcceptedBeforeDedup int `json:"accepted_before_dedup"` AcceptedAfterDedup int `json:"accepted_after_dedup"` EnteredAI int `json:"entered_ai"` + FetchDurationMS int64 `json:"fetch_duration_ms,omitempty"` + ResponseBytes int64 `json:"response_bytes,omitempty"` + CacheStatus string `json:"cache_status,omitempty"` } type SourceStatsError struct { @@ -103,6 +108,8 @@ func (report *SourceStatsReport) RecalculateTotals() { totals.AcceptedBeforeDedup += source.AcceptedBeforeDedup totals.AcceptedAfterDedup += source.AcceptedAfterDedup totals.EnteredAI += source.EnteredAI + totals.FetchDurationMS += source.FetchDurationMS + totals.ResponseBytes += source.ResponseBytes } report.Totals = totals } diff --git a/internal/watch/announcement_site.go b/internal/watch/announcement_site.go index 5f644c8..39b601a 100644 --- a/internal/watch/announcement_site.go +++ b/internal/watch/announcement_site.go @@ -33,7 +33,7 @@ var claudeReleaseNotesMonthByName = map[string]time.Month{ "december": time.December, } -func runAnnouncementSite(ctx context.Context, site config.WatchSite, now time.Time, indexState IndexState, articleState ArticleState, fetchHTML fetchHTMLFunc, articleConcurrency int) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { +func runAnnouncementSite(ctx context.Context, site config.WatchSite, now time.Time, indexState IndexState, articleState ArticleState, fetchHTML fetchHTMLFunc, articleConcurrency int, deepVerifyInterval time.Duration, deepVerifyBatchSize int) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { homeHTML, err := fetchHTML(ctx, site.HomeURL) if err != nil { return nil, nil, nil, err @@ -66,14 +66,16 @@ func runAnnouncementSite(ctx context.Context, site config.WatchSite, now time.Ti } return runWatchCategory(ctx, watchCategoryRun{ - site: site, - now: now, - stateKey: stateKey, - current: current, - indexState: indexState, - articleState: articleState, - fetchContent: fetchContent, - articleConcurrency: articleConcurrency, + site: site, + now: now, + stateKey: stateKey, + current: current, + indexState: indexState, + articleState: articleState, + fetchContent: fetchContent, + articleConcurrency: articleConcurrency, + deepVerifyInterval: deepVerifyInterval, + deepVerifyBatchSize: deepVerifyBatchSize, }) } diff --git a/internal/watch/run.go b/internal/watch/run.go index 4aaff5b..ab3bcf4 100644 --- a/internal/watch/run.go +++ b/internal/watch/run.go @@ -200,7 +200,7 @@ func runContext(ctx context.Context, cfg *config.Config, now time.Time, fetchHTM if err := ctx.Err(); err != nil { return nil, nil, err } - siteArticles, siteSeenItems, events, err := runSite(ctx, site, now, indexState, articleState, fetchHTML, articleConcurrency) + siteArticles, siteSeenItems, events, err := runSite(ctx, site, now, indexState, articleState, fetchHTML, articleConcurrency, cfg.Watch.DeepVerifyInterval, cfg.Watch.DeepVerifyBatchSize) if err != nil { if ctxErr := ctx.Err(); ctxErr != nil { return nil, nil, ctxErr @@ -235,18 +235,18 @@ func runContext(ctx context.Context, cfg *config.Config, now time.Time, fetchHTM return articles, report, nil } -func runSite(ctx context.Context, site config.WatchSite, now time.Time, indexState IndexState, articleState ArticleState, fetchHTML fetchHTMLFunc, articleConcurrency int) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { +func runSite(ctx context.Context, site config.WatchSite, now time.Time, indexState IndexState, articleState ArticleState, fetchHTML fetchHTMLFunc, articleConcurrency int, deepVerifyInterval time.Duration, deepVerifyBatchSize int) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { switch site.Type { case config.WatchTypeAnthropicSupport: - return runAnthropicSupportSite(ctx, site, now, indexState, articleState, fetchHTML, articleConcurrency) + return runAnthropicSupportSite(ctx, site, now, indexState, articleState, fetchHTML, articleConcurrency, deepVerifyInterval, deepVerifyBatchSize) case config.WatchTypeAnnouncementPage: - return runAnnouncementSite(ctx, site, now, indexState, articleState, fetchHTML, articleConcurrency) + return runAnnouncementSite(ctx, site, now, indexState, articleState, fetchHTML, articleConcurrency, deepVerifyInterval, deepVerifyBatchSize) default: return nil, nil, nil, nil } } -func runAnthropicSupportSite(ctx context.Context, site config.WatchSite, now time.Time, indexState IndexState, articleState ArticleState, fetchHTML fetchHTMLFunc, articleConcurrency int) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { +func runAnthropicSupportSite(ctx context.Context, site config.WatchSite, now time.Time, indexState IndexState, articleState ArticleState, fetchHTML fetchHTMLFunc, articleConcurrency int, deepVerifyInterval time.Duration, deepVerifyBatchSize int) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { homeHTML, err := fetchHTML(ctx, site.HomeURL) if err != nil { return nil, nil, nil, err @@ -297,14 +297,16 @@ func runAnthropicSupportSite(ctx context.Context, site config.WatchSite, now tim current.SnapshotAt = now categoryArticles, categorySeenItems, categoryEvents, err := runWatchCategory(ctx, watchCategoryRun{ - site: site, - now: now, - stateKey: watchCategoryStateKey(site.Name, current.Category), - current: current, - indexState: indexState, - articleState: articleState, - fetchContent: fetchContent, - articleConcurrency: articleConcurrency, + site: site, + now: now, + stateKey: watchCategoryStateKey(site.Name, current.Category), + current: current, + indexState: indexState, + articleState: articleState, + fetchContent: fetchContent, + articleConcurrency: articleConcurrency, + deepVerifyInterval: deepVerifyInterval, + deepVerifyBatchSize: deepVerifyBatchSize, }) if err != nil { return nil, nil, nil, err diff --git a/internal/watch/run_test.go b/internal/watch/run_test.go index d088af5..3741517 100644 --- a/internal/watch/run_test.go +++ b/internal/watch/run_test.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "net/http/httptest" + "slices" "strings" "sync" "testing" @@ -669,7 +670,7 @@ func TestRunBackfillsMissingArticleStateForExistingCategoryBaseline(t *testing.T responses["https://support.claude.com/zh-CN/articles/14328960-claude-上的-身份验证"] = `Claude 上的身份验证

Claude 上的身份验证

某些使用场景需要提供政府颁发的身份证件与实时自拍。

新增了实时自拍与手机号码交叉校验。

` - articles, report, err = runContext(context.Background(), cfg, time.Date(2026, 4, 15, 17, 0, 0, 0, time.UTC), fetchHTML) + articles, report, err = runContext(context.Background(), cfg, time.Date(2026, 4, 16, 17, 0, 0, 0, time.UTC), fetchHTML) if err != nil { t.Fatalf("Run() second error = %v", err) } @@ -1092,7 +1093,7 @@ func TestRunContentChangedUpdatesSeenState(t *testing.T) { } } -func TestRunWatchCategoryChecksExistingArticlesConcurrently(t *testing.T) { +func TestRunWatchCategoryDeepVerifiesOldestDueArticlesConcurrently(t *testing.T) { now := time.Date(2026, 6, 14, 12, 0, 0, 0, time.UTC) items := []model.WatchIndexItem{ {Title: "Article 1", URL: "https://example.com/1", ItemHash: "1"}, @@ -1105,10 +1106,11 @@ func TestRunWatchCategoryChecksExistingArticlesConcurrently(t *testing.T) { for _, item := range items { titles[item.URL] = item.Title articleState[item.URL] = model.WatchArticleState{ - URL: item.URL, - Title: item.Title, - SummaryHash: hashWatchContent("summary"), - BodyHash: hashWatchContent("body"), + URL: item.URL, + Title: item.Title, + SummaryHash: hashWatchContent("summary"), + BodyHash: hashWatchContent("body"), + LastCheckedAt: now.Add(-48 * time.Hour), } } indexState := IndexState{Categories: map[string]model.WatchIndexSnapshot{ @@ -1138,14 +1140,16 @@ func TestRunWatchCategoryChecksExistingArticlesConcurrently(t *testing.T) { } articles, seenItems, events, err := runWatchCategory(context.Background(), watchCategoryRun{ - site: config.WatchSite{Name: "source"}, - now: now, - stateKey: "source::category", - current: indexState.Categories["source::category"], - indexState: indexState, - articleState: articleState, - fetchContent: fetchContent, - articleConcurrency: 2, + site: config.WatchSite{Name: "source"}, + now: now, + stateKey: "source::category", + current: indexState.Categories["source::category"], + indexState: indexState, + articleState: articleState, + fetchContent: fetchContent, + articleConcurrency: 2, + deepVerifyInterval: 24 * time.Hour, + deepVerifyBatchSize: len(items), }) if err != nil { t.Fatalf("runWatchCategory() error = %v", err) @@ -1160,3 +1164,69 @@ func TestRunWatchCategoryChecksExistingArticlesConcurrently(t *testing.T) { t.Fatalf("maxInFlight = %d, want configured concurrency 2", maxInFlight) } } + +func TestRunWatchCategorySkipsFreshArticleBodies(t *testing.T) { + now := time.Date(2026, 6, 14, 12, 0, 0, 0, time.UTC) + item := model.WatchIndexItem{Title: "Article", URL: "https://example.com/1", ItemHash: "1"} + indexState := IndexState{Categories: map[string]model.WatchIndexSnapshot{ + "source::category": {Category: "category", ItemCount: 1, Items: []model.WatchIndexItem{item}}, + }} + articleState := ArticleState{item.URL: { + URL: item.URL, Title: item.Title, + SummaryHash: hashWatchContent("summary"), BodyHash: hashWatchContent("body"), + LastCheckedAt: now.Add(-time.Hour), + }} + calls := 0 + _, _, events, err := runWatchCategory(context.Background(), watchCategoryRun{ + site: config.WatchSite{Name: "source"}, now: now, stateKey: "source::category", + current: indexState.Categories["source::category"], indexState: indexState, + articleState: articleState, articleConcurrency: 2, + deepVerifyInterval: 24 * time.Hour, deepVerifyBatchSize: 48, + fetchContent: func(context.Context, string) (watchArticleContent, error) { + calls++ + return watchArticleContent{}, nil + }, + }) + if err != nil { + t.Fatalf("runWatchCategory() error = %v", err) + } + if calls != 0 || len(events) != 0 { + t.Fatalf("calls=%d events=%#v, want fast index-only path", calls, events) + } +} + +func TestRunWatchCategoryBoundsDeepVerificationBatch(t *testing.T) { + now := time.Date(2026, 6, 14, 12, 0, 0, 0, time.UTC) + items := []model.WatchIndexItem{ + {Title: "oldest", URL: "https://example.com/1", Position: 1, ItemHash: "1"}, + {Title: "middle", URL: "https://example.com/2", Position: 2, ItemHash: "2"}, + {Title: "newest", URL: "https://example.com/3", Position: 3, ItemHash: "3"}, + } + articleState := ArticleState{} + for i, item := range items { + articleState[item.URL] = model.WatchArticleState{URL: item.URL, Title: item.Title, + SummaryHash: hashWatchContent("summary"), BodyHash: hashWatchContent("body"), + LastCheckedAt: now.Add(time.Duration(i-4) * 24 * time.Hour)} + } + indexState := IndexState{Categories: map[string]model.WatchIndexSnapshot{ + "source::category": {Category: "category", ItemCount: len(items), Items: items}, + }} + fetched := []string{} + _, _, _, err := runWatchCategory(context.Background(), watchCategoryRun{ + site: config.WatchSite{Name: "source"}, now: now, stateKey: "source::category", + current: indexState.Categories["source::category"], indexState: indexState, + articleState: articleState, articleConcurrency: 1, + deepVerifyInterval: 24 * time.Hour, deepVerifyBatchSize: 2, + fetchContent: func(_ context.Context, url string) (watchArticleContent, error) { + fetched = append(fetched, url) + state := articleState[url] + return watchArticleContent{title: state.Title, summary: "summary", body: "body"}, nil + }, + }) + if err != nil { + t.Fatalf("runWatchCategory() error = %v", err) + } + if !slices.Equal(fetched, []string{items[0].URL, items[1].URL}) { + t.Fatalf("fetched = %#v, want oldest two", fetched) + } +} diff --git a/internal/watch/site_article_state.go b/internal/watch/site_article_state.go index 9e9f3de..6a3220e 100644 --- a/internal/watch/site_article_state.go +++ b/internal/watch/site_article_state.go @@ -67,14 +67,16 @@ type watchArticleContentResult struct { } type watchCategoryRun struct { - site config.WatchSite - now time.Time - stateKey string - current model.WatchIndexSnapshot - indexState IndexState - articleState ArticleState - fetchContent watchArticleContentFetcher - articleConcurrency int + site config.WatchSite + now time.Time + stateKey string + current model.WatchIndexSnapshot + indexState IndexState + articleState ArticleState + fetchContent watchArticleContentFetcher + articleConcurrency int + deepVerifyInterval time.Duration + deepVerifyBatchSize int } func runWatchCategory(ctx context.Context, run watchCategoryRun) ([]model.Article, []model.WatchSeenArticle, []model.WatchEvent, error) { @@ -197,12 +199,43 @@ func updateCurrentWatchArticleStates(ctx context.Context, run watchCategoryRun, for _, url := range changedURLs { changed[url] = struct{}{} } - items := make([]model.WatchIndexItem, 0, len(run.current.Items)) + interval := run.deepVerifyInterval + if interval <= 0 { + interval = config.DefaultWatchDeepVerifyInterval + } + batchSize := run.deepVerifyBatchSize + if batchSize < 1 { + batchSize = config.DefaultWatchDeepVerifyBatchSize + } + type deepVerifyCandidate struct { + item model.WatchIndexItem + lastChecked time.Time + } + candidates := make([]deepVerifyCandidate, 0, len(run.current.Items)) for _, item := range run.current.Items { if _, ok := changed[item.URL]; ok { continue } - items = append(items, item) + state, ok := run.articleState[item.URL] + if !ok || state.LastCheckedAt.IsZero() || !run.now.Before(state.LastCheckedAt.Add(interval)) { + candidates = append(candidates, deepVerifyCandidate{item: item, lastChecked: state.LastCheckedAt}) + } + } + slices.SortStableFunc(candidates, func(a, b deepVerifyCandidate) int { + if a.lastChecked.Equal(b.lastChecked) { + return a.item.Position - b.item.Position + } + if a.lastChecked.Before(b.lastChecked) { + return -1 + } + return 1 + }) + if len(candidates) > batchSize { + candidates = candidates[:batchSize] + } + items := make([]model.WatchIndexItem, len(candidates)) + for i := range candidates { + items[i] = candidates[i].item } results, err := fetchWatchArticleContents(ctx, items, run.fetchContent, run.articleConcurrency) if err != nil {