import hashlib
import json
from contextlib import closing
from pathlib import Path
import sqlite3
import tempfile
import unittest
from unittest.mock import patch

from paper_atlas.collectors.proceedings import discover, parse_paper
from paper_atlas.collectors.dblp import XML_SHA
from paper_atlas.collectors.dblp_records import identifiers, normalize_row, collect
from paper_atlas.collectors.dblp_scope import scope
from paper_atlas.collectors.public_openreview import normalize
from paper_atlas.db.submissions import identities_for_row
from paper_atlas.ingestion.records import ingest_file
from paper_atlas.io.publication_ids import mlsys_id


HASH = '0123456789abcdef' * 2
ROOT = 'https://proceedings.mlsys.org/'


def page(year=2026, modern=True):
    return ROOT + ('paper_files/paper/' if modern else 'paper/') + str(year) + '/hash/' + HASH + ('-Abstract-Conference.html' if modern else '-Abstract.html')


def raw_row(family='mm', book='ACM Multimedia', year=2024):
    return {'record_type': 'inproceedings', 'key': f'conf/{family}/Paper24', 'publtype': None,
            'fields': {'title': ['A published paper'], 'author': ['Jane Doe', 'John Smith'],
                       'booktitle': [book], 'year': [str(year)], 'crossref': [f'conf/{family}/{year}']}}


def volume(family='mm', year=2024, title=None):
    return {'record_type': 'proceedings', 'key': f'conf/{family}/{year}',
            'fields': {'title': [title or f'ACM Multimedia {year}'], 'year': [str(year)]}}


class MlsysMultimediaTests(unittest.TestCase):
    def test_mlsys_discovery_retains_all_official_years_and_ignores_pdf_or_other_host(self):
        html = ''.join(f'<a href="/paper_files/paper/{year}">MLSys {year}</a>' for year in range(2019, 2027))
        html += '<a href="https://example.org/paper_files/paper/2018">Other</a><a href="/paper/2026/file/paper.pdf">PDF</a>'
        with patch('paper_atlas.collectors.proceedings.fetch', return_value=html.encode()):
            self.assertEqual([year for _, year in discover('mlsys', '/unused')], list(range(2019, 2027)))

    def test_mlsys_metadata_has_stable_publisher_identity_and_explicit_track(self):
        html = '<meta name="citation_title" content="A title"><meta name="citation_author" content="Doe, Jane"><span class="paper-track">Conference</span><h2>Abstract</h2><p>Real abstract.</p><h4>Name Change Policy</h4><p>Footer.</p>'
        row = parse_paper(html, page(), 'mlsys', 2026, ROOT + 'paper_files/paper/2026')
        self.assertEqual((row['venue'], row['track'], row['abstract'], row['authors']), ('MLSys', 'main', 'Real abstract.', ['Jane Doe']))
        self.assertEqual(row['identifiers']['publication_mlsys'], '2026/' + HASH)
        with self.assertRaisesRegex(ValueError, 'Unclassified'):
            parse_paper(html.replace('Conference', 'New Unknown Track'), page(), 'mlsys', 2026, ROOT)
        with self.assertRaisesRegex(ValueError, 'publication year'):
            parse_paper(html, page(), 'mlsys', 2025, ROOT)
        with self.assertRaisesRegex(ValueError, 'non-archival'):
            parse_paper(html, page(2018), 'mlsys', 2018, ROOT)

    def test_year_label_and_retained_bibliography_links_bridge_the_2023_format(self):
        url = page(2023).replace('-Conference.html', '-mlsys2023.html')
        self.assertEqual(mlsys_id(url), '2023/' + HASH)
        self.assertIsNone(mlsys_id(url.replace('mlsys2023.html', 'mlsys2022.html')))
        official = parse_paper('<h1>Updated title</h1><span class="paper-track">mlsys2023</span><p class="paper-abstract">Official abstract</p>', url, 'mlsys', 2023, ROOT)
        self.assertEqual(official['track'], 'main')
        legacy = {'source': 'dblp', 'external_id': 'conf/mlsys/paper23', 'url': 'https://dblp.org/rec/conf/mlsys/paper23',
                  'venue': 'MLSys', 'year': 2023, 'track': 'main', 'status': 'accepted', 'title': 'Bibliographic title',
                  'abstract': None, 'authors': [], 'identifiers': {}, 'metadata': {'bibliography': {'ee': [url]}}}
        self.assertEqual(identities_for_row(legacy), [('publication_mlsys', '2023/' + HASH)])
        self.assertEqual(identities_for_row({**legacy, 'status': 'unknown'}), [])
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); db = root / 'papers.sqlite'
            for row in [legacy, official]:
                path = root / (row['source'] + '.jsonl'); path.write_text(json.dumps(row) + '\n')
                path.with_suffix('.manifest.json').write_text(json.dumps({'source': row['source'], 'parser_version': 2, 'list_url': ROOT, 'count': 1}))
                ingest_file(db, path)
            with closing(sqlite3.connect(db)) as c:
                self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
                self.assertEqual(c.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 2)

    def test_legacy_mlsys_and_sysml_alias_preserve_same_publication_identity(self):
        self.assertEqual(mlsys_id(page(2019, False)), mlsys_id(page(2019)))
        self.assertIsNone(mlsys_id(page().replace('proceedings.mlsys.org', 'example.org')))
        own, ambiguous = identifiers({'ee': [page(2019, False), page(2019)], 'cite': [page(2025)]})
        self.assertEqual(own, {'publication_mlsys': '2019/' + HASH}); self.assertFalse(ambiguous)
        _, ambiguous = identifiers({'ee': [page(2019), page(2019).replace(HASH, 'a' * 32)]})
        self.assertEqual(len(ambiguous['publication_mlsys']), 2)
        row = parse_paper('<h1>Legacy</h1><p class="paper-abstract">Abstract</p>', page(2019, False), 'mlsys', 2019, ROOT)
        self.assertEqual(row['venue_aliases'], ['SysML'])
        self.assertEqual(row['track'], 'main')

    def test_mm_main_numbered_parts_and_workshops_remain_separate(self):
        for book in ['ACM Multimedia', 'ACM Multimedia (1)', 'ACM Multimedia (2)']:
            result, reason = scope(raw_row(book=book), volume())
            self.assertIsNone(reason); self.assertEqual(result['venue'], 'MM'); self.assertEqual(result['track'], 'proceedings_unspecified')
            self.assertIn('per-paper section', result['track_scope_note'])
            self.assertIn('ACMMM', result['venue_aliases'])
        self.assertEqual(scope(raw_row(book='ACM Multimedia Workshops'), volume())[0]['track'], 'workshop')
        self.assertEqual(scope(raw_row(book='ACM Multimedia (Thematic Workshops)'), volume())[0]['track'], 'workshop')
        self.assertIsNone(scope(raw_row(book='MMSports@MM'), volume())[0])
        self.assertEqual(normalize_row(raw_row(book='ACM Multimedia Workshops'), {'conf/mm/2024': volume()}, {'MM'})[1], 'separate_workshop')

    def test_sysml_alias_requires_machine_learning_proceedings(self):
        row = raw_row('mlsys', 'SysML', 2019)
        result, reason = scope(row, volume('mlsys', 2019, 'Conference on Machine Learning and Systems, SysML 2019'))
        self.assertIsNone(reason); self.assertEqual(result['venue'], 'MLSys'); self.assertEqual(result['venue_aliases'], ['SysML'])
        self.assertIsNone(scope(row, volume('mlsys', 2019, 'Systems Modeling Language 2019'))[0])

    def test_bibliography_and_official_abstract_import_as_one_paper_and_submission(self):
        raw = raw_row('mlsys', 'SysML', 2019); raw['fields']['ee'] = [page(2019, False)]
        bib, reason = normalize_row(raw, {'conf/mlsys/2019': volume('mlsys', 2019, 'Machine Learning and Systems 2019')}, {'MLSys'})
        self.assertIsNone(reason)
        official = parse_paper('<meta name="citation_title" content="An updated title"><p class="paper-abstract">The official abstract.</p>', page(2019), 'mlsys', 2019, ROOT)
        # Different title and absent official authors deliberately prevent a title-based bridge.
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); db = root / 'papers.sqlite'
            for row in [bib, official]:
                path = root / (row['source'] + '.jsonl'); path.write_text(json.dumps(row) + '\n')
                path.with_suffix('.manifest.json').write_text(json.dumps({'source': row['source'], 'parser_version': 2,
                    'list_url': ROOT, 'count': 1, 'venue': 'MLSys', 'year': 2019, 'track': 'main'}))
                ingest_file(db, path)
            with closing(sqlite3.connect(db)) as c:
                self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
                self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
                self.assertEqual(c.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 2)
                self.assertEqual(c.execute('SELECT abstract FROM papers').fetchone()[0], 'The official abstract.')
                self.assertEqual(c.execute('PRAGMA foreign_key_check').fetchall(), [])

    def test_publisher_identity_cannot_bridge_rejection_other_year_or_venue(self):
        row = {'source': 'mlsys', 'venue': 'MLSys', 'year': 2026, 'status': 'accepted',
               'identifiers': {'publication_mlsys': '2026/' + HASH}}
        self.assertEqual(identities_for_row(row), [('publication_mlsys', '2026/' + HASH)])
        self.assertEqual(identities_for_row({**row, 'status': 'rejected'}), [])
        with self.assertRaisesRegex(ValueError, 'Invalid authoritative'):
            identities_for_row({**row, 'identifiers': {}})
        for changed in [dict(year=2025), dict(venue='MM'), dict(source='curated')]:
            bad = {**row, **changed, 'submission_identifiers': [{'scheme': 'publication_mlsys', 'value': '2026/' + HASH}]}
            with self.assertRaisesRegex(ValueError, 'Invalid authoritative'):
                identities_for_row(bad)

    def test_explicit_mm_negative_assertion_maps_to_mm_without_inventing_labels(self):
        provenance = {'source_file': 'https://example.org/pinned', 'source_line': 1}
        row = {'id': 'realForum1', 'title': 'A title', 'abstract': 'An abstract', 'track': 'main', 'status': 'Reject'}
        normalized, reason = normalize(row, 'acmmm', 2025, provenance)
        self.assertIsNone(reason); self.assertEqual(normalized['venue'], 'MM')
        self.assertEqual(normalized['status'], 'rejected')
        self.assertIsNone(normalize({**row, 'status': 'Submitted'}, 'acmmm', 2025, provenance)[0])

    def test_old_candidate_subset_cannot_silently_omit_new_requested_venues(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); catalogue = root / 'dblp-catalogue.sqlite'
            with closing(sqlite3.connect(catalogue)) as c, c:
                c.execute('CREATE TABLE proceedings(key TEXT PRIMARY KEY,metadata_json TEXT)')
                c.execute('INSERT INTO proceedings VALUES(?,?)', ('conf/mm/2024', json.dumps(volume())))
            candidates = root / 'candidates.jsonl'; candidates.write_text(json.dumps(raw_row()) + '\n')
            original = candidates.read_bytes()
            catalogue.with_name('dblp-ai-candidates.manifest.json').write_text(json.dumps({
                'snapshot_sha256': XML_SHA, 'candidate_sha256': hashlib.sha256(original).hexdigest(),
                'candidate_family_counts': {'aaai': 1}, 'record_types': {'proceedings': 1}}))
            with self.assertRaisesRegex(ValueError, 'original --archive'):
                collect(catalogue, root / 'out', ['MM'], candidates=candidates)
            self.assertEqual(candidates.read_bytes(), original)
            self.assertFalse((root / 'out').exists())


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