import gzip
import hashlib
import json
from pathlib import Path
import sqlite3
import tempfile
import unittest

from paper_atlas.collectors.dblp import catalogue,records


class DblpCatalogueTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup);self.root=Path(self.tmp.name)
        self.dtd=self.root/'dblp.dtd';self.dtd.write_bytes(b'<!ENTITY eacute "&#233;">')

    def archive(self,body,doctype='<!DOCTYPE dblp SYSTEM "dblp.dtd">'):
        path=self.root/'dblp.xml.gz'
        with gzip.open(path,'wb') as stream:stream.write(('<?xml version="1.0"?>'+doctype+'<dblp>'+body+'</dblp>').encode())
        return path

    def test_streamed_entities_nested_titles_and_candidate_scope_remain_explicit(self):
        path=self.archive('<proceedings key="conf/aaai/2025"><title>AAAI Workshop proceedings</title><year>2025</year></proceedings><inproceedings key="conf/aaai/paper"><author>Ren&eacute;</author><title>Learn <i>deep</i> models</title><year>2025</year><booktitle>AAAI Workshop</booktitle><crossref>conf/aaai/2025</crossref><ee>https://doi.org/10.1234/example</ee></inproceedings><inproceedings key="conf/unrelated/paper"><title>Other</title><year>2020</year></inproceedings>')
        manifest=catalogue(path,self.dtd,self.root/'out',hashlib.sha256(path.read_bytes()).hexdigest(),hashlib.md5(self.dtd.read_bytes()).hexdigest())
        self.assertEqual(manifest['records_scanned'],3);self.assertEqual(manifest['candidate_records'],1)
        row=json.loads(Path(manifest['file']).read_text())
        self.assertEqual(row['fields']['author'],['René']);self.assertEqual(row['fields']['title'],['Learn deep models'])
        self.assertNotIn('status',row);self.assertNotIn('venue',row)
        with sqlite3.connect(manifest['catalogue']) as connection:
            self.assertEqual(connection.execute('SELECT sum(count) FROM volume_counts').fetchone()[0],2)
            self.assertEqual(connection.execute('SELECT count(*) FROM proceedings').fetchone()[0],1)

    def test_remote_external_dtd_is_rejected_without_fetching(self):
        path=self.archive('',doctype='<!DOCTYPE dblp SYSTEM "https://example.org/evil.dtd">')
        with self.assertRaisesRegex(ValueError,'pinned external'):list(records(path,self.dtd.read_bytes()))

    def test_real_monthly_dtd_name_and_latin1_encoding(self):
        path=self.root/'monthly.xml.gz'
        xml='<?xml version="1.0" encoding="ISO-8859-1"?><!DOCTYPE dblp SYSTEM "dblp-2023-06-28.dtd"><dblp><inproceedings key="conf/aaai/one"><author>René</author><title>Metadata</title></inproceedings></dblp>'
        with gzip.open(path,'wb') as stream:stream.write(xml.encode('latin1'))
        self.assertEqual(list(records(path,self.dtd.read_bytes()))[0]['fields']['author'],['René'])

    def test_multiple_volume_fields_remain_reviewable_without_double_counting(self):
        path=self.archive('<inproceedings key="conf/aaai/joint"><title>Joint publication</title><year>2024</year><booktitle>First meeting</booktitle><booktitle>Second meeting</booktitle><crossref>conf/aaai/joint2024</crossref></inproceedings>')
        manifest=catalogue(path,self.dtd,self.root/'out',hashlib.sha256(path.read_bytes()).hexdigest(),hashlib.md5(self.dtd.read_bytes()).hexdigest())
        with sqlite3.connect(manifest['catalogue']) as connection:
            raw=json.loads(connection.execute('SELECT metadata_json FROM volume_anomalies').fetchone()[0])
            self.assertEqual(raw['fields']['booktitle'],['First meeting','Second meeting'])
            self.assertEqual(connection.execute('SELECT sum(count) FROM volume_counts').fetchone()[0],1)

    def test_internal_entities_are_rejected(self):
        path=self.archive('',doctype='<!DOCTYPE dblp SYSTEM "dblp.dtd" [<!ENTITY x "expanded">]>')
        with self.assertRaisesRegex(ValueError,'pinned external'):list(records(path,self.dtd.read_bytes()))

    def test_checksum_failure_preserves_existing_output(self):
        path=self.archive('');out=self.root/'out';out.mkdir();target=out/'dblp-ai-candidates.jsonl';target.write_text('previous')
        with self.assertRaisesRegex(ValueError,'snapshot checksum'):catalogue(path,self.dtd,out,'0'*64)
        self.assertEqual(target.read_text(),'previous')

    def test_truncated_gzip_cannot_publish_a_catalogue(self):
        path=self.archive('<inproceedings key="conf/aaai/one"><title>One</title></inproceedings>')
        path.write_bytes(path.read_bytes()[:-8]);out=self.root/'out'
        with self.assertRaises((EOFError,OSError)):
            catalogue(path,self.dtd,out,hashlib.sha256(path.read_bytes()).hexdigest(),hashlib.md5(self.dtd.read_bytes()).hexdigest())
        self.assertFalse((out/'dblp-ai-candidates.jsonl').exists())


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