"""Clustering behaviour, using synthetic vectors so no model download is needed."""

import numpy as np
import pytest

from news import cluster, store
from news.util import stable_id


@pytest.fixture
def con(tmp_path):
    c = store.connect(tmp_path / "t.db")
    store.upsert_source(
        c, id="s1", name="A", url="http://a", kind="rss", bucket="a",
        weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
    store.upsert_source(
        c, id="s2", name="B", url="http://b", kind="rss", bucket="b",
        weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
    return c


def add(con, item_id, source_id, when, chash):
    con.execute(
        """INSERT INTO items (id,source_id,title,summary,url,canonical_url,
                              published_at,first_seen_at,content_hash)
           VALUES (?,?,?,'',?,?,?,?,?)""",
        (item_id, source_id, f"t-{item_id}", f"http://x/{item_id}",
         f"http://x/{item_id}", when, when, chash))
    return con.execute("SELECT * FROM items WHERE id=?", (item_id,)).fetchone()


def unit(*v):
    a = np.array(v, dtype=np.float32)
    return a / np.linalg.norm(a)


def test_similar_items_merge_and_distinct_ones_do_not(con):
    near_a, near_b = unit(1, 0.02, 0), unit(1, 0.06, 0)   # cosine ~0.999
    far = unit(0, 1, 0)                                    # orthogonal
    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1"),
            add(con, "i2", "s2", "2026-01-01T01:00:00+00:00", "h2"),
            add(con, "i3", "s1", "2026-01-01T02:00:00+00:00", "h3")]
    out = cluster.assign(con, rows, {"h1": near_a, "h2": near_b, "h3": far})
    by_item = {a.item_id: a.cluster_id for a in out}
    assert by_item["i1"] == by_item["i2"], "paraphrase-near items must merge"
    assert by_item["i3"] != by_item["i1"], "unrelated item must not merge"


def test_items_outside_the_window_do_not_merge(con):
    v = unit(1, 0, 0)
    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1"),
            add(con, "i2", "s2", "2026-01-05T00:00:00+00:00", "h2")]  # 4 days later
    out = cluster.assign(con, rows, {"h1": v, "h2": v})
    assert out[0].cluster_id != out[1].cluster_id


def test_bucket_count_counts_newsrooms_not_feeds(con):
    """Two feeds from one newsroom are one independent confirmation, not two."""
    v = unit(1, 0, 0)
    store.upsert_source(
        con, id="s3", name="A2", url="http://a2", kind="rss", bucket="a",
        weight=1.0, prominence=0, enabled=1, poll_interval_secs=900)
    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1"),
            add(con, "i2", "s3", "2026-01-01T01:00:00+00:00", "h2")]
    out = cluster.assign(con, rows, {"h1": v, "h2": v})
    assert out[0].cluster_id == out[1].cluster_id
    assert cluster.bucket_count(con, out[0].cluster_id) == 1

    rows2 = [add(con, "i3", "s2", "2026-01-01T02:00:00+00:00", "h3")]
    cluster.assign(con, rows2, {"h3": v})
    assert cluster.bucket_count(con, out[0].cluster_id) == 2


def test_assignment_is_idempotent(con):
    v = unit(1, 0, 0)
    rows = [add(con, "i1", "s1", "2026-01-01T00:00:00+00:00", "h1")]
    a = cluster.assign(con, rows, {"h1": v})
    b = cluster.assign(con, rows, {"h1": v})
    assert a[0].cluster_id == b[0].cluster_id
    n = con.execute("SELECT COUNT(*) c FROM cluster_members").fetchone()["c"]
    assert n == 1, "re-assigning an item must not duplicate membership"


def test_syndicated_partners_in_one_bucket_count_once(con):
    """Regression: Ohio Capital Journal and Kentucky Lantern are both States
    Newsroom and carried 10 identical headlines in one poll. Configured as
    separate buckets they would have doubled the corroboration signal."""
    v = unit(1, 0, 0)
    for sid, name in (("sn1", "Ohio Capital Journal"), ("sn2", "Kentucky Lantern")):
        store.upsert_source(con, id=sid, name=name, url=f"http://{sid}", kind="rss",
                            bucket="states-newsroom", weight=1.0, prominence=0,
                            enabled=1, poll_interval_secs=900)
    rows = [add(con, "s1i", "sn1", "2026-01-01T00:00:00+00:00", "hs1"),
            add(con, "s2i", "sn2", "2026-01-01T00:30:00+00:00", "hs2")]
    out = cluster.assign(con, rows, {"hs1": v, "hs2": v})
    assert out[0].cluster_id == out[1].cluster_id
    assert cluster.bucket_count(con, out[0].cluster_id) == 1, (
        "syndicated partners must not read as independent confirmation")


def _cluster_of(con, items_spec, vec):
    """Put every item in one cluster and return its id."""
    rows = [add(con, iid, sid, when, chash) for iid, sid, when, chash in items_spec]
    out = cluster.assign(con, rows, {c: vec for *_, c in items_spec})
    return out[0].cluster_id


def test_origin_count_collapses_syndicated_copy_across_newsrooms(con):
    """Different owners, identical text: a wire reprint is not corroboration.

    This is the case no static bucket can express -- a local station is
    independent on local stories and a pure reprinter on national ones.
    """
    v = unit(1, 0, 0)
    con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
                ("hw", 3, store.to_blob(v)))
    rows = [add(con, "w1", "s1", "2026-01-01T00:00:00+00:00", "hw"),
            add(con, "w2", "s2", "2026-01-01T00:20:00+00:00", "hw")]
    # identical headline text across two different buckets
    con.execute("UPDATE items SET title='Wire dispatch: minister resigns' WHERE id IN ('w1','w2')")
    rows = con.execute("SELECT * FROM items WHERE id IN ('w1','w2') ORDER BY published_at").fetchall()
    cid = cluster.assign(con, list(rows), {"hw": v})[0].cluster_id
    assert cluster.bucket_count(con, cid) == 2
    assert cluster.origin_count(con, cid) == 1


def test_origin_count_keeps_independently_written_reports_separate(con):
    """Same event, different newsrooms, different wording: real corroboration."""
    # cosine ~0.90: above the 0.82 clustering threshold so they land in one
    # cluster, below the 0.97 syndication threshold so they stay distinct
    # origins. The first draft of this test used ~0.999 vectors and correctly
    # merged them -- the test data was wrong, not the rule.
    a, b = unit(1, 0, 0), unit(0.90, 0.436, 0)
    for h, v in (("ha", a), ("hb", b)):
        con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
                    (h, 3, store.to_blob(v)))
    add(con, "r1", "s1", "2026-01-01T00:00:00+00:00", "ha")
    add(con, "r2", "s2", "2026-01-01T00:30:00+00:00", "hb")
    con.execute("UPDATE items SET title='Thieves steal two paintings' WHERE id='r1'")
    con.execute("UPDATE items SET title='Renoir works taken in museum heist' WHERE id='r2'")
    rows = con.execute("SELECT * FROM items WHERE id IN ('r1','r2') ORDER BY published_at").fetchall()
    cid = cluster.assign(con, list(rows), {"ha": a, "hb": b})[0].cluster_id
    assert cluster.origin_count(con, cid) == 2


def test_origin_count_never_exceeds_bucket_count(con):
    """Syndication detection may only ever reduce corroboration."""
    v = unit(1, 0, 0)
    con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
                ("hx", 3, store.to_blob(v)))
    specs = [("x1", "s1", "2026-01-01T00:00:00+00:00", "hx"),
             ("x2", "s2", "2026-01-01T00:10:00+00:00", "hx"),
             ("x3", "s1", "2026-01-01T00:20:00+00:00", "hx")]
    cid = _cluster_of(con, specs, v)
    assert cluster.origin_count(con, cid) <= cluster.bucket_count(con, cid)


def equicos(n, c=0.90, dim=None):
    """n unit vectors with pairwise cosine exactly `c`.

    Needed because these tests must sit in a narrow band: above the 0.82
    clustering threshold so the items form ONE cluster, below the 0.97
    syndication threshold so they read as distinct pieces of writing. Earlier
    drafts of this test used ad-hoc vectors that fell outside it in both
    directions and tested nothing.
    """
    dim = dim or n + 1
    out = []
    for i in range(n):
        v = np.zeros(dim, dtype=np.float32)
        v[0] = np.sqrt(c)
        v[1 + i] = np.sqrt(1 - c)
        out.append(v)
    return out


def _mk(con, spec, vec_for):
    """spec: list of (item_id, source_id, title, content_hash). One cluster."""
    when = "2026-01-01T00:{:02d}:00+00:00"
    rows = []
    for n, (iid, sid, title, chash) in enumerate(spec):
        con.execute(
            """INSERT INTO items (id,source_id,title,summary,url,canonical_url,
                                  published_at,first_seen_at,content_hash)
               VALUES (?,?,?,'',?,?,?,?,?)""",
            (iid, sid, title, f"http://x/{iid}", f"http://x/{iid}",
             when.format(n), when.format(n), chash))
        v = vec_for(chash)
        con.execute("INSERT OR REPLACE INTO embeddings (content_hash,dim,vec) VALUES (?,?,?)",
                    (chash, len(v), store.to_blob(v)))
    rows = con.execute("SELECT * FROM items ORDER BY published_at").fetchall()
    v = {r["content_hash"]: vec_for(r["content_hash"]) for r in rows}
    return cluster.assign(con, list(rows), v)[0].cluster_id


def _sources(con, *pairs):
    for sid, bucket in pairs:
        store.upsert_source(con, id=sid, name=sid, url=f"http://{sid}", kind="rss",
                            bucket=bucket, weight=1.0, prominence=0, enabled=1,
                            poll_interval_secs=900)


def test_corroboration_ranks_the_four_cases_in_the_right_order(con):
    """The 2x2 that motivates the split, each case scored in isolation.

    Ordering must be: four independent newsrooms > four outlets running one
    wire story > one newsroom filing four pieces > one story on two feeds of
    the same newsroom.
    """
    # "wire" is one shared vector (cosine 1.0 -> syndicated); t0..t3 and u0..u3
    # are pairwise 0.90 -> one cluster, four distinct pieces of writing.
    distinct = equicos(4, 0.90, dim=6)
    table = {"wire": np.eye(6, dtype=np.float32)[0]}
    for i in range(4):
        table[f"t{i}"] = distinct[i]
        table[f"u{i}"] = distinct[i]

    def vec(h):
        return table[h]

    # (a) one newsroom, one story, two feeds -> pure duplicate
    c1 = store.connect(":memory:")
    _sources(c1, ("f1", "bbc"), ("f2", "bbc"))
    cid = _mk(c1, [("a1", "f1", "Same headline", "wire"),
                   ("a2", "f2", "Same headline", "wire")], vec)
    dup = cluster.corroboration(c1, cid)

    # (b) one newsroom, four distinct pieces -> depth
    c2 = store.connect(":memory:")
    _sources(c2, ("g1", "nyt"))
    cid = _mk(c2, [(f"b{i}", "g1", f"Distinct angle {i}", f"t{i}") for i in range(4)], vec)
    deep = cluster.corroboration(c2, cid)

    # (c) four newsrooms, one syndicated wire story -> pickups
    c3 = store.connect(":memory:")
    _sources(c3, *[(f"h{i}", f"local{i}") for i in range(4)])
    cid = _mk(c3, [(f"c{i}", f"h{i}", "Wire dispatch text", "wire") for i in range(4)], vec)
    wire = cluster.corroboration(c3, cid)

    # (d) four newsrooms each writing their own -> origins
    c4 = store.connect(":memory:")
    _sources(c4, *[(f"k{i}", f"nat{i}") for i in range(4)])
    cid = _mk(c4, [(f"d{i}", f"k{i}", f"Own reporting {i}", f"u{i}") for i in range(4)], vec)
    orig = cluster.corroboration(c4, cid)

    assert dup.origins == 1 and dup.pickups == 0 and dup.depth == 0
    assert deep.origins == 1 and deep.depth == 3
    assert wire.origins == 1 and wire.pickups == 3
    assert orig.origins == 4

    assert orig.score > wire.score > deep.score > dup.score, (
        f"expected origins > pickups > depth > duplicate, got "
        f"{orig.score:.2f} / {wire.score:.2f} / {deep.score:.2f} / {dup.score:.2f}")


def test_origins_dominate_the_tiebreakers(con):
    """No amount of depth or pickups may outweigh one extra real newsroom."""
    c = store.connect(":memory:")
    _sources(c, ("m1", "one"))
    vs = equicos(20, 0.90)
    table = {f"v{i}": vs[i] for i in range(20)}
    cid = _mk(c, [(f"e{i}", "m1", f"Piece {i}", f"v{i}") for i in range(20)],
              lambda h: table[h])
    twenty_deep = cluster.corroboration(c, cid)
    assert twenty_deep.score < 2.0, (
        "20 pieces from one newsroom must still rank below two independent ones")