diff --git a/skills/last30days/scripts/store.py b/skills/last30days/scripts/store.py index 39f8286..ed3dbc3 100644 --- a/skills/last30days/scripts/store.py +++ b/skills/last30days/scripts/store.py @@ -360,6 +360,22 @@ def update_run(run_id: int, **kwargs): conn.close() +def get_latest_completed_runs(topic_id: int, limit: int = 2) -> List[Dict[str, Any]]: + """Return newest completed runs for a topic.""" + conn = _connect() + try: + rows = conn.execute( + """SELECT * FROM research_runs + WHERE topic_id = ? AND status = 'completed' + ORDER BY datetime(run_date) DESC, id DESC + LIMIT ?""", + (topic_id, limit), + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + # --- Findings --- @@ -536,6 +552,84 @@ def get_sightings_for_run(topic_id: int, run_id: int) -> List[Dict[str, Any]]: conn.close() +def compute_topic_delta(topic_id: int) -> Dict[str, Any]: + """Compare the latest completed watchlist run with the previous run.""" + runs = get_latest_completed_runs(topic_id, limit=2) + topic = _get_topic_by_id(topic_id) + topic_name = topic["name"] if topic else str(topic_id) + if len(runs) < 2: + return { + "topic": topic_name, + "status": "insufficient_history", + "message": "Need at least two completed runs to compute a delta.", + } + + current_run, previous_run = runs[0], runs[1] + current = _sightings_by_url(get_sightings_for_run(topic_id, current_run["id"])) + previous = _sightings_by_url(get_sightings_for_run(topic_id, previous_run["id"])) + + current_urls = set(current) + previous_urls = set(previous) + new_urls = sorted(current_urls - previous_urls) + continued_urls = sorted(current_urls & previous_urls) + dropped_urls = sorted(previous_urls - current_urls) + + findings = { + "new": [current[url] for url in new_urls], + "continued": [current[url] for url in continued_urls], + "dropped": [previous[url] for url in dropped_urls], + } + + return { + "topic": topic_name, + "status": "ok", + "current_run_id": current_run["id"], + "previous_run_id": previous_run["id"], + "new": len(new_urls), + "continued": len(continued_urls), + "dropped": len(dropped_urls), + "sources": _delta_source_counts(findings), + "findings": findings, + } + + +def _get_topic_by_id(topic_id: int) -> Optional[Dict[str, Any]]: + conn = _connect() + try: + row = conn.execute("SELECT * FROM topics WHERE id = ?", (topic_id,)).fetchone() + return dict(row) if row else None + finally: + conn.close() + + +def _sightings_by_url(sightings: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: + return { + sighting["source_url"]: sighting + for sighting in sightings + if sighting.get("source_url") + } + + +def _delta_source_counts( + findings: Dict[str, List[Dict[str, Any]]] +) -> Dict[str, Dict[str, int]]: + sources = sorted({ + finding.get("source") or "unknown" + for group in findings.values() + for finding in group + }) + counts = { + source: {"new": 0, "continued": 0, "dropped": 0} + for source in sources + } + for group_name, group in findings.items(): + for finding in group: + source = finding.get("source") or "unknown" + counts.setdefault(source, {"new": 0, "continued": 0, "dropped": 0}) + counts[source][group_name] += 1 + return counts + + def get_new_findings( topic_id: int, since: Optional[str] = None, diff --git a/skills/last30days/scripts/watchlist.py b/skills/last30days/scripts/watchlist.py index e991fb4..bf2903a 100644 --- a/skills/last30days/scripts/watchlist.py +++ b/skills/last30days/scripts/watchlist.py @@ -111,6 +111,14 @@ def cmd_list(args): }, default=str)) +def cmd_delta(args): + topic = store.get_topic(args.topic) + if not topic: + print(json.dumps({"error": f'Topic not found: "{args.topic}"'})) + sys.exit(1) + print(json.dumps(store.compute_topic_delta(topic["id"]), default=str)) + + def cmd_run_one(args): topic = store.get_topic(args.topic) if not topic: @@ -252,6 +260,10 @@ def build_parser() -> argparse.ArgumentParser: list_parser = sub.add_parser("list") list_parser.set_defaults(func=cmd_list) + delta = sub.add_parser("delta") + delta.add_argument("topic") + delta.set_defaults(func=cmd_delta) + run_one = sub.add_parser("run-one") run_one.add_argument("topic") run_one.set_defaults(func=cmd_run_one) diff --git a/tests/test_store.py b/tests/test_store.py index f5c1eb3..b35755a 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -565,6 +565,88 @@ def test_store_findings_updates_existing_sighting_for_same_run(temp_db): assert sightings[0]["source_title"] == "Reddit 1 updated" assert sightings[0]["engagement_score"] == 15.0 +def test_get_latest_completed_runs_returns_newest_completed_only(temp_db): + """Test latest-run lookup ignores failed runs and orders newest first.""" + topic = store.add_topic("Test Topic") + old_run_id = store.record_run(topic["id"], source_mode="v3", status="completed") + store.record_run(topic["id"], source_mode="v3", status="failed") + latest_run_id = store.record_run(topic["id"], source_mode="v3", status="completed") + + runs = store.get_latest_completed_runs(topic["id"], limit=2) + + assert [run["id"] for run in runs] == [latest_run_id, old_run_id] + + +def test_compute_topic_delta_compares_latest_two_runs(temp_db): + """Test watchlist delta classification using per-run sightings.""" + topic = store.add_topic("Test Topic") + previous_run_id = store.record_run(topic["id"], source_mode="v3", status="completed") + store.store_findings(previous_run_id, topic["id"], [ + { + "source": "reddit", + "source_url": "https://reddit.com/continued", + "source_title": "Continued", + "content": "Still present", + "engagement_score": 10.0, + "relevance_score": 0.7, + }, + { + "source": "x", + "source_url": "https://x.com/dropped/status/1", + "source_title": "Dropped", + "content": "Dropped this run", + "engagement_score": 20.0, + "relevance_score": 0.8, + }, + ]) + current_run_id = store.record_run(topic["id"], source_mode="v3", status="completed") + store.store_findings(current_run_id, topic["id"], [ + { + "source": "reddit", + "source_url": "https://reddit.com/continued", + "source_title": "Continued", + "content": "Still present", + "engagement_score": 15.0, + "relevance_score": 0.7, + }, + { + "source": "github", + "source_url": "https://github.com/example/new", + "source_title": "New", + "content": "New this run", + "engagement_score": 30.0, + "relevance_score": 0.9, + }, + ]) + + delta = store.compute_topic_delta(topic["id"]) + + assert delta["status"] == "ok" + assert delta["current_run_id"] == current_run_id + assert delta["previous_run_id"] == previous_run_id + assert delta["new"] == 1 + assert delta["continued"] == 1 + assert delta["dropped"] == 1 + assert [f["source_url"] for f in delta["findings"]["new"]] == ["https://github.com/example/new"] + assert [f["source_url"] for f in delta["findings"]["continued"]] == ["https://reddit.com/continued"] + assert [f["source_url"] for f in delta["findings"]["dropped"]] == ["https://x.com/dropped/status/1"] + assert delta["sources"] == { + "github": {"new": 1, "continued": 0, "dropped": 0}, + "reddit": {"new": 0, "continued": 1, "dropped": 0}, + "x": {"new": 0, "continued": 0, "dropped": 1}, + } + + +def test_compute_topic_delta_requires_two_completed_runs(temp_db): + """Test that delta reports insufficient history before two successful runs.""" + topic = store.add_topic("Test Topic") + store.record_run(topic["id"], source_mode="v3", status="completed") + + delta = store.compute_topic_delta(topic["id"]) + + assert delta["status"] == "insufficient_history" + assert "Need at least two completed runs" in delta["message"] + def test_update_validates_allowed_columns(temp_db, sample_report): """Test update_run/update_finding accept valid keys and reject invalid keys.""" diff --git a/tests/test_watchlist_commands.py b/tests/test_watchlist_commands.py index 28ff488..eef8334 100644 --- a/tests/test_watchlist_commands.py +++ b/tests/test_watchlist_commands.py @@ -177,6 +177,61 @@ def test_cmd_list_with_topics(temp_db, capsys): assert topic_names == {"Topic 1", "Topic 2", "Topic 3"} +# === Tests for cmd_delta() === + +def test_cmd_delta_outputs_topic_delta(temp_db, capsys): + """Test printing the latest watchlist delta as JSON.""" + topic = store.add_topic("Test Topic") + previous_run_id = store.record_run(topic["id"], source_mode="v3", status="completed") + store.store_findings(previous_run_id, topic["id"], [ + { + "source": "reddit", + "source_url": "https://reddit.com/continued", + "source_title": "Continued", + "content": "Still present", + } + ]) + current_run_id = store.record_run(topic["id"], source_mode="v3", status="completed") + store.store_findings(current_run_id, topic["id"], [ + { + "source": "reddit", + "source_url": "https://reddit.com/continued", + "source_title": "Continued", + "content": "Still present", + }, + { + "source": "github", + "source_url": "https://github.com/example/new", + "source_title": "New", + "content": "New this run", + }, + ]) + + args = Mock() + args.topic = "Test Topic" + + watchlist.cmd_delta(args) + + captured = capsys.readouterr() + output = json.loads(captured.out) + + assert output["topic"] == "Test Topic" + assert output["status"] == "ok" + assert output["current_run_id"] == current_run_id + assert output["previous_run_id"] == previous_run_id + assert output["new"] == 1 + assert output["continued"] == 1 + + +def test_cmd_delta_unknown_topic_exits(temp_db): + """Test delta for an unknown topic exits with an error.""" + args = Mock() + args.topic = "Missing Topic" + + with pytest.raises(SystemExit): + watchlist.cmd_delta(args) + + # === Tests for cmd_config() === def test_cmd_config_delivery(temp_db, capsys):