import json
from contextlib import closing
import sqlite3
import tempfile
import unittest
import zipfile
from pathlib import Path

from paper_atlas.ingestion.arxiv import ingest
from paper_atlas.db.store import Store
from paper_atlas.ingestion.arxiv import refresh_categories


class ImportTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.db = self.root/'papers.sqlite'
        self.archive = self.root/'metadata.zip'

    def write(self, rows):
        with zipfile.ZipFile(self.archive,'w') as z:
            z.writestr('arxiv-metadata-oai-snapshot.json','\n'.join(json.dumps(r) for r in rows))

    def row(self, identifier, doi=None):
        return dict(id=identifier,title='A shared title',abstract='Searchable abstract',authors_parsed=[['Doe','Jane','']],categories='cs.AI cs.LG',doi=doi,versions=[])

    def test_identity_provenance_and_idempotency(self):
        self.write([self.row('2601.00001','10.0000/example'),self.row('2601.00002','10.0000/example')])
        ingest(self.db,self.archive)
        ingest(self.db,self.archive)
        with closing(sqlite3.connect(self.db)) as c, c:
            self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
            self.assertEqual(c.execute('SELECT count(*) FROM source_records').fetchone()[0],2)
            self.assertEqual(c.execute("SELECT count(*) FROM papers_fts WHERE papers_fts MATCH 'Searchable'").fetchone()[0],1)
            self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],0)
            self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(),[])
            self.assertEqual(c.execute('PRAGMA integrity_check').fetchone()[0],'ok')

    def test_title_alone_does_not_merge(self):
        self.write([self.row('2601.00001'),self.row('2601.00002')])
        ingest(self.db,self.archive)
        with closing(sqlite3.connect(self.db)) as c, c:
            self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],2)

    def test_platform_and_custom_tags_remain_distinct(self):
        self.write([self.row('2601.00001')])
        ingest(self.db,self.archive)
        with closing(sqlite3.connect(self.db)) as c, c:
            paper = c.execute('SELECT id FROM papers').fetchone()[0]
            tag = c.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','cs.AI')").lastrowid
            c.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)',(paper,tag,'manual:1'))
        ingest(self.db,self.archive)
        with closing(sqlite3.connect(self.db)) as c, c:
            self.assertEqual(c.execute("SELECT count(*) FROM paper_tags pt JOIN tags t ON pt.tag_id=t.id WHERE t.value='cs.AI'").fetchone()[0],2)

    def test_merged_records_retain_separate_category_evidence_across_refresh(self):
        one=self.row('2601.00001','10.0000/example');two=self.row('2601.00002','10.0000/example')
        self.write([one,two]);ingest(self.db,self.archive)
        with closing(sqlite3.connect(self.db)) as c:
            self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],4)
            self.assertEqual(c.execute('SELECT count(DISTINCT source_record_id) FROM paper_tags').fetchone()[0],2)
        one['categories']='cs.AI';two['categories']='cs.LG';self.write([one,two]);ingest(self.db,self.archive)
        with closing(sqlite3.connect(self.db)) as c:
            actual=c.execute('SELECT r.external_id,t.value FROM paper_tags pt JOIN source_records r ON r.id=pt.source_record_id JOIN tags t ON t.id=pt.tag_id ORDER BY r.external_id').fetchall()
            self.assertEqual(actual,[('2601.00001','cs.AI'),('2601.00002','cs.LG')])

    def test_migrate_legacy_categories_without_removing_custom_evidence(self):
        store=Store(self.db);self.addCleanup(store.conn.close)
        paper,record=store.record('arxiv','2601.00001','https://arxiv.org/abs/2601.00001','A title','AB',[])
        for namespace,key in [('arxiv:category','arxiv:categories'),('custom:topic','manual:1')]:
            tag=store.conn.execute('INSERT INTO tags(namespace,value) VALUES(?,?)',(namespace,'cs.AI')).lastrowid
            store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,key))
        refresh_categories(store,paper,record,'cs.AI cs.LG');refresh_categories(store,paper,record,'cs.AI cs.LG')
        self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0],3)
        self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='arxiv:categories'").fetchone()[0],0)
        self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='manual:1'").fetchone()[0],1)


if __name__ == '__main__':
    unittest.main()