228
return isinstance(window, dict) and window.get("since") == iso_timestamp(since) and window.get("until") == iso_timestamp(until)
229
230
231
+def same_utc_day(value: str | None, expected: date) -> bool:
232
+ parsed = parse_iso_datetime(value)
233
+ return parsed is not None and parsed.astimezone(UTC).date() == expected
234
+
235
+
236
+def source_reuse_decisions(
237
+ payload: dict[str, Any] | None,
238
+ sources: list[NewsSourceConfig],
239
+ *,
240
+ week: str,
241
+ run_date: date,
242
+ since: datetime,
243
+ until: datetime,
244
+ policy: str,
245
+ current_config_checksum: str,
246
+ current_code_sha: str | None,
247
+) -> tuple[list[dict[str, Any]], list[NewsSourceConfig], list[dict[str, Any]], list[dict[str, Any]]]:
248
+ """Compatibility planner for callers that pass a loaded artifact."""
249
+ decisions: list[dict[str, Any]] = []
250
+ reused_articles: list[dict[str, Any]] = []
251
+ reused_statuses: list[dict[str, Any]] = []
252
+ to_crawl: list[NewsSourceConfig] = []
253
+ metadata = payload.get("metadata", {}) if isinstance(payload, dict) else {}
254
+ statuses = metadata.get("source_status", []) if isinstance(metadata, dict) else []
255
+ if not isinstance(statuses, list):
256
+ statuses = []
257
+ status_by_source = {str(status.get("source")): status for status in statuses if isinstance(status, dict)}
258
+ raw_articles = payload.get("articles", []) if isinstance(payload, dict) else []
259
+ articles: list[dict[str, Any]] = []
260
+ articles_malformed = False
261
+ if isinstance(raw_articles, list):
262
+ for article in raw_articles:
263
+ article_sources = article.get("sources", []) if isinstance(article, dict) else None
264
+ if not isinstance(article, dict) or not isinstance(article_sources, list):
265
+ articles_malformed = True
266
+ break
267
+ articles.append(article)
268
+ else:
269
+ articles_malformed = True
270
+ global_reasons: list[str] = []
271
+ if policy == "force-refresh":
272
+ global_reasons.append("source_refresh_policy=force-refresh")
273
+ if payload is None:
274
+ global_reasons.append("artifact missing or malformed")
275
+ else:
276
+ if payload.get("week") != week:
277
+ global_reasons.append(f"week mismatch: expected {week}, found {payload.get('week')!r}")
278
+ if not same_utc_day(payload.get("crawled_at"), run_date):
279
+ global_reasons.append("artifact is not from the current UTC run date")
280
+ window = payload.get("crawl_window") if isinstance(payload.get("crawl_window"), dict) else {}
281
+ if window.get("since") != iso_timestamp(since) or window.get("until") != iso_timestamp(until):
282
+ global_reasons.append("crawl window mismatch")
283
+ if isinstance(metadata, dict):
284
+ if metadata.get("source_config_checksum") != current_config_checksum:
285
+ global_reasons.append("source config checksum mismatch")
286
+ artifact_code_sha = metadata.get("crawler_code_sha")
287
+ if current_code_sha and artifact_code_sha != current_code_sha:
288
+ global_reasons.append("crawler/config fingerprint mismatch")
289
+ if articles_malformed:
290
+ global_reasons.append("artifact articles malformed")
291
+ for source in sources:
292
+ source_reasons = list(global_reasons)
293
+ status = status_by_source.get(source.name)
294
+ if not status or status.get("success") is not True:
295
+ source_reasons.append("source missing or previously failed")
296
+ if source_reasons:
297
+ decisions.append({"source": source.name, "decision": "refresh", "reasons": source_reasons})
298
+ to_crawl.append(source)
299
+ continue
300
+ source_articles = [
301
+ article for article in articles
302
+ if source.name in {str(article.get("source", "")), *[str(item) for item in article.get("sources", [])]}
303
+ ]
304
+ reused_articles.extend(source_articles)
305
+ reused_status = dict(status)
306
+ reused_status["reused_same_day"] = True
307
+ reused_status["success"] = True
308
+ reused_statuses.append(reused_status)
309
+ decisions.append({"source": source.name, "decision": "reuse", "reasons": []})
310
+ return reused_articles, to_crawl, reused_statuses, decisions
311
+
312
+
313
def plan_source_reuse(
314
previous_path: Path,
315
sources: list[NewsSourceConfig],
319
until: datetime,
320
config_checksum: str,
321
forced_sources: set[str] | None = None,
322
+ source_refresh_policy: str = "reuse-same-day",
323
+ run_started_at: datetime | None = None,
324
+ current_code_sha: str | None = None,
325
) -> tuple[list[dict[str, Any]], list[NewsSourceConfig], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, str]]]:
326
"""Load eligible same-day source artifacts and return reused articles plus sources to crawl."""
327
forced = forced_sources or set()
328
+ run_time = run_started_at or now
329
requested = {source.name for source in sources}
330
pending: list[NewsSourceConfig] = []
331
reused_articles: list[dict[str, Any]] = []
335
previous = _load_json_object(previous_path) if previous_path.exists() else None
336
expected_schema_checksum = schema_checksum()
337
252
- if previous is None:
338
+ if source_refresh_policy == "force-refresh":
339
+ stale_reasons = ["source_refresh_policy=force-refresh"]
340
+ elif previous is None:
341
stale_reasons = ["missing previous artifact" if not previous_path.exists() else "previous artifact is not valid JSON"]
342
else:
343
crawled_at = parse_iso_datetime(previous.get("crawled_at"))
348
stale_reasons.append(str(exc))
349
if previous.get("week") != week_slug(now):
350
stale_reasons.append(f"week mismatch: expected {week_slug(now)}, found {previous.get('week')!r}")
263
- if crawled_at is None or crawled_at.astimezone(UTC).date() != now.astimezone(UTC).date():
351
+ if crawled_at is None or crawled_at.astimezone(UTC).date() != run_time.astimezone(UTC).date():
352
stale_reasons.append("crawled_at is not from the current UTC day")
353
if not _same_window(previous, since, until):
354
stale_reasons.append("crawl_window mismatch")
356
stale_reasons.append("source_config_checksum mismatch")
357
if metadata.get("schema_checksum") != expected_schema_checksum:
358
stale_reasons.append("schema_checksum mismatch")
359
+ artifact_code_sha = metadata.get("crawler_code_sha")
360
+ if current_code_sha and artifact_code_sha != current_code_sha:
361
+ stale_reasons.append("crawler/config fingerprint mismatch")
362
363
previous_metadata = previous.get("metadata", {}) if isinstance(previous, dict) and isinstance(previous.get("metadata"), dict) else {}
364
previous_statuses = {
1010
default=[],
1011
help="Refresh one source by id even when its same-day artifact is reusable. Can be repeated.",
1012
)
1013
+ parser.add_argument(
1014
+ "--reuse-artifact",
1015
+ default=None,
1016
+ help="Existing external-news artifact to reuse per source when fresh for this run window.",
1017
+ )
1018
+ parser.add_argument(
1019
+ "--source-refresh-policy",
1020
+ choices=["reuse-same-day", "refresh-missing-stale", "force-refresh"],
1021
+ default="reuse-same-day",
1022
+ help="Source refresh policy for reruns (default: reuse eligible same-day sources).",
1023
+ )
1024
+ parser.add_argument(
1025
+ "--run-started-at",
1026
+ default=None,
1027
+ help="UTC run start timestamp used for same-day reuse checks (ISO 8601). Defaults to now.",
1028
+ )
1029
+ parser.add_argument(
1030
+ "--current-code-sha",
1031
+ default=None,
1032
+ help="Optional crawler/config fingerprint; reused artifacts with a conflicting fingerprint are stale.",
1033
+ )
1034
args = parser.parse_args(argv)
1035
1036
now = datetime.now(UTC)
1037
+ run_started_at = parse_iso_datetime(args.run_started_at) if args.run_started_at else now
1038
+ if run_started_at is None:
1039
+ print("--run-started-at must be an ISO 8601 timestamp", file=sys.stderr)
1040
+ return 1
1041
since = (
1042
datetime.strptime(args.since, "%Y-%m-%d").replace(tzinfo=UTC)
1043
if args.since
1058
out_path = out_dir / f"{week_slug(now)}-external-news.json"
1059
1060
config_checksum = source_config_checksum(source_configs)
945
- force_sources = {source.name for source in source_configs} if args.force_refresh else set(args.force_refresh_source or [])
1061
+ source_refresh_policy = "force-refresh" if args.force_refresh else args.source_refresh_policy
1062
+ current_code_sha = args.current_code_sha or ""
1063
+ force_sources = {source.name for source in source_configs} if source_refresh_policy == "force-refresh" else set(args.force_refresh_source or [])
1064
+ reuse_path = Path(args.reuse_artifact) if args.reuse_artifact else out_path
1065
reused_articles, sources_to_crawl, reuse_summary, provenance, _ = plan_source_reuse(
947
- out_path,
1066
+ reuse_path,
1067
source_configs,
1068
now=now,
1069
since=since,
1070
until=until,
1071
config_checksum=config_checksum,
1072
forced_sources=force_sources,
1073
+ source_refresh_policy=source_refresh_policy,
1074
+ run_started_at=run_started_at,
1075
+ current_code_sha=current_code_sha,
1076
)
1077
refreshed_articles, errors, refreshed_statuses = crawl_sources_parallel(
1078
sources_to_crawl, since=since, until=until, max_workers=args.max_workers
1091
run_id=os.environ.get("GITHUB_RUN_ID", "local"),
1092
)
1093
reused_sources = {entry["source_id"] for entry in provenance if entry.get("action") == "reused"}
972
- previous = _load_json_object(out_path) if out_path.exists() else None
1094
+ previous = _load_json_object(reuse_path) if reuse_path.exists() else None
1095
previous_metadata = previous.get("metadata", {}) if isinstance(previous, dict) and isinstance(previous.get("metadata"), dict) else {}
1096
previous_statuses = [
1097
{**status, "reused": True}
1116
source_artifact_provenance=provenance,
1117
run_id=os.environ.get("GITHUB_RUN_ID", "local"),
1118
)
1119
+ output["metadata"]["same_day_reuse"] = (
1120
+ "mixed" if reused_articles and refreshed_articles else "reused" if reused_articles else "not_reused"
1121
+ )
1122
+ output["metadata"]["source_refresh_policy"] = source_refresh_policy
1123
+ output["metadata"]["source_reuse_decisions"] = [
1124
+ {"source": item["source"], "decision": "reuse" if item["action"] == "reused" else "refresh", "reasons": item["reasons"]}
1125
+ for item in output["metadata"]["source_reuse_summary"]
1126
+ ]
1127
+ output["metadata"]["crawler_code_sha"] = current_code_sha
1128
+ output["metadata"]["artifact_checksum"] = artifact_checksum(output)
1129
+ validate_canonical_output(output)
1130
1131
out_path.parent.mkdir(parents=True, exist_ok=True)
1132
with open(out_path, "w", encoding="utf-8") as f: