fix(watchlist): refresh sighting retries
This commit is contained in:
@@ -446,7 +446,7 @@ def store_findings(
|
|||||||
|
|
||||||
new_count = len(insert_rows)
|
new_count = len(insert_rows)
|
||||||
updated_count = len(update_rows)
|
updated_count = len(update_rows)
|
||||||
_record_sightings(conn, run_id, topic_id, with_urls)
|
_record_sightings(conn, run_id, topic_id, with_urls, existing_by_url)
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?",
|
"UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?",
|
||||||
(new_count, updated_count, run_id),
|
(new_count, updated_count, run_id),
|
||||||
@@ -463,6 +463,7 @@ def _record_sightings(
|
|||||||
run_id: int,
|
run_id: int,
|
||||||
topic_id: int,
|
topic_id: int,
|
||||||
findings_with_urls: List[tuple[str, Dict[str, Any]]],
|
findings_with_urls: List[tuple[str, Dict[str, Any]]],
|
||||||
|
existing_by_url: Optional[Dict[str, sqlite3.Row]] = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Record the findings observed during this run.
|
"""Record the findings observed during this run.
|
||||||
|
|
||||||
@@ -474,20 +475,28 @@ def _record_sightings(
|
|||||||
return
|
return
|
||||||
|
|
||||||
by_url = {url: finding for url, finding in findings_with_urls}
|
by_url = {url: finding for url, finding in findings_with_urls}
|
||||||
placeholders = ",".join("?" for _ in by_url)
|
rows_by_url = dict(existing_by_url or {})
|
||||||
rows = conn.execute(
|
|
||||||
f"SELECT id, source_url FROM findings WHERE source_url IN ({placeholders})",
|
missing_urls = [url for url in by_url if url not in rows_by_url]
|
||||||
list(by_url),
|
if missing_urls:
|
||||||
).fetchall()
|
placeholders = ",".join("?" for _ in missing_urls)
|
||||||
|
rows = conn.execute(
|
||||||
|
f"SELECT id, source_url FROM findings WHERE source_url IN ({placeholders})",
|
||||||
|
missing_urls,
|
||||||
|
).fetchall()
|
||||||
|
rows_by_url.update({row["source_url"]: row for row in rows})
|
||||||
|
|
||||||
sighting_rows = []
|
sighting_rows = []
|
||||||
for row in rows:
|
for url, finding in by_url.items():
|
||||||
finding = by_url[row["source_url"]]
|
row = rows_by_url.get(url)
|
||||||
|
if row is None:
|
||||||
|
continue
|
||||||
sighting_rows.append((
|
sighting_rows.append((
|
||||||
row["id"],
|
row["id"],
|
||||||
run_id,
|
run_id,
|
||||||
topic_id,
|
topic_id,
|
||||||
finding.get("source", "unknown"),
|
finding.get("source", "unknown"),
|
||||||
row["source_url"],
|
url,
|
||||||
finding.get("source_title") or finding.get("title", ""),
|
finding.get("source_title") or finding.get("title", ""),
|
||||||
finding.get("engagement_score", 0),
|
finding.get("engagement_score", 0),
|
||||||
finding.get("relevance_score", 0),
|
finding.get("relevance_score", 0),
|
||||||
@@ -497,10 +506,17 @@ def _record_sightings(
|
|||||||
return
|
return
|
||||||
|
|
||||||
conn.executemany(
|
conn.executemany(
|
||||||
"""INSERT OR IGNORE INTO finding_sightings
|
"""INSERT INTO finding_sightings
|
||||||
(finding_id, run_id, topic_id, source, source_url, source_title,
|
(finding_id, run_id, topic_id, source, source_url, source_title,
|
||||||
engagement_score, relevance_score)
|
engagement_score, relevance_score)
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
ON CONFLICT(run_id, finding_id) DO UPDATE SET
|
||||||
|
topic_id = excluded.topic_id,
|
||||||
|
source = excluded.source,
|
||||||
|
source_url = excluded.source_url,
|
||||||
|
source_title = excluded.source_title,
|
||||||
|
engagement_score = excluded.engagement_score,
|
||||||
|
relevance_score = excluded.relevance_score""",
|
||||||
sighting_rows,
|
sighting_rows,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -535,6 +535,32 @@ def test_store_findings_sightings_are_idempotent_per_run(temp_db):
|
|||||||
assert len(sightings) == 1
|
assert len(sightings) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_store_findings_updates_existing_sighting_for_same_run(temp_db):
|
||||||
|
"""Test that retrying a run refreshes its sighting snapshot instead of freezing it."""
|
||||||
|
topic = store.add_topic("Test Topic")
|
||||||
|
run_id = store.record_run(topic["id"], source_mode="v3")
|
||||||
|
finding = {
|
||||||
|
"source": "reddit",
|
||||||
|
"source_url": "https://reddit.com/1",
|
||||||
|
"source_title": "Reddit 1",
|
||||||
|
"content": "Content",
|
||||||
|
"engagement_score": 10.0,
|
||||||
|
"relevance_score": 0.7,
|
||||||
|
}
|
||||||
|
|
||||||
|
store.store_findings(run_id, topic["id"], [finding])
|
||||||
|
store.store_findings(
|
||||||
|
run_id,
|
||||||
|
topic["id"],
|
||||||
|
[{**finding, "source_title": "Reddit 1 updated", "engagement_score": 15.0}],
|
||||||
|
)
|
||||||
|
|
||||||
|
sightings = store.get_sightings_for_run(topic["id"], run_id)
|
||||||
|
assert len(sightings) == 1
|
||||||
|
assert sightings[0]["source_title"] == "Reddit 1 updated"
|
||||||
|
assert sightings[0]["engagement_score"] == 15.0
|
||||||
|
|
||||||
|
|
||||||
def test_update_validates_allowed_columns(temp_db, sample_report):
|
def test_update_validates_allowed_columns(temp_db, sample_report):
|
||||||
"""Test update_run/update_finding accept valid keys and reject invalid keys."""
|
"""Test update_run/update_finding accept valid keys and reject invalid keys."""
|
||||||
topic = store.add_topic("Test Topic")
|
topic = store.add_topic("Test Topic")
|
||||||
|
|||||||
Reference in New Issue
Block a user