"""Immutable JUnit ingestion and retry-aware, local CI result analysis.

The database belongs to one repository and comparison stream. Caller-supplied
sequence numbers define build order; filesystem timestamps do not.
"""
from __future__ import annotations
import argparse
import base64
import csv
import hashlib
import io
import json
import math
import os
from pathlib import Path
import re
import sqlite3
import sys
import tempfile
import xml.etree.ElementTree as ET

APPLICATION_ID = 0x43544931
MAX_BYTES = 10 * 1024 * 1024
MAX_CASES = 50000
MAX_ELEMENTS = 300000
MAX_DEPTH = 64
MAX_TEXT = 32768
FAILED = {'failure', 'error'}
STATUSES = ['new failure', 'repeated failure', 'failure without baseline', 'retry pass', 'recovered', 'pass', 'skipped', 'not reported']

class InvalidReport(ValueError):
    pass

class Conflict(ValueError):
    pass


def identity(value, field):
    if not isinstance(value, str) or not value.strip() or len(value) > 160 or any(ord(c) < 32 for c in value):
        raise InvalidReport(f'{field} must be a nonempty string of at most 160 printable characters')
    return value


def positive_int(value, field):
    if type(value) is not int or value < 1 or value > 2147483647:
        raise InvalidReport(f'{field} must be an integer from 1 to 2147483647')
    return value


def local_name(tag):
    return tag.rsplit('}', 1)[-1]


def parse_xml(content: bytes, filename='report.xml'):
    """Parse a bounded UTF-8 JUnit subset. Never expand DTD/entity declarations."""
    if len(content) > MAX_BYTES:
        raise InvalidReport(f'{filename}: exceeds {MAX_BYTES} byte limit')
    try:
        text = content.decode('utf-8-sig')
    except UnicodeDecodeError as exc:
        raise InvalidReport(f'{filename}: UTF-8 XML required') from exc
    if '\x00' in text or re.search(r'<!\s*(DOCTYPE|ENTITY)', text, re.I):
        raise InvalidReport(f'{filename}: DTD, entities and NUL bytes are not accepted')
    declaration = re.match(r'\s*<\?xml\b[^?]*encoding\s*=\s*[\"\']([^\"\']+)', text, re.I)
    if declaration and declaration[1].lower().replace('-', '') != 'utf8':
        raise InvalidReport(f'{filename}: XML declaration must use UTF-8')
    try:
        root = ET.fromstring(text)
    except ET.ParseError as exc:
        raise InvalidReport(f'{filename}: invalid XML: {exc}') from exc
    if local_name(root.tag) not in ('testsuite', 'testsuites'):
        raise InvalidReport(f'{filename}: root must be testsuite or testsuites')
    stack = [(root, 1)]
    count = 0
    cases = []
    seen = set()
    while stack:
        node, depth = stack.pop()
        count += 1
        if depth > MAX_DEPTH or count > MAX_ELEMENTS:
            raise InvalidReport(f'{filename}: XML structure limit exceeded')
        if any(len(k) > 256 or len(v) > 4096 for k, v in node.attrib.items()):
            raise InvalidReport(f'{filename}: attribute length limit exceeded')
        if local_name(node.tag) == 'testcase':
            name = node.get('name', '')
            if not name:
                raise InvalidReport(f'{filename}: testcase is missing name')
            key_fields = [node.get('file', ''), node.get('classname', ''), name]
            key = json.dumps(key_fields, separators=(',', ':'), ensure_ascii=True)
            if key in seen:
                raise InvalidReport(f'{filename}: duplicate test identity {key}')
            seen.add(key)
            raw_time = node.get('time')
            try:
                duration = None if raw_time in (None, '') else float(raw_time)
            except ValueError as exc:
                raise InvalidReport(f'{filename}: invalid test duration') from exc
            if duration is not None and (not math.isfinite(duration) or duration < 0 or duration > 604800):
                raise InvalidReport(f'{filename}: duration must be finite, nonnegative and at most seven days')
            outcome = [child for child in node if local_name(child.tag) in ('failure', 'error', 'skipped')]
            if len(outcome) > 1:
                raise InvalidReport(f'{filename}: ambiguous multiple outcome elements for {key}')
            status = local_name(outcome[0].tag) if outcome else 'pass'
            message = ''
            if outcome:
                item = outcome[0]
                message = (item.get('message', '') + '\n' + ''.join(item.itertext())).strip()
            truncated = len(message) > MAX_TEXT
            if truncated:
                message = message[:MAX_TEXT] + '\n[Message truncated; original XML retained.]'
            cases.append(dict(key=key, file=key_fields[0], classname=key_fields[1], name=name,
                              status=status, duration=duration, message=message,
                              message_truncated=truncated, artifact=filename))
            if len(cases) > MAX_CASES:
                raise InvalidReport(f'{filename}: testcase count limit exceeded')
        stack.extend((child, depth + 1) for child in reversed(list(node)))
    if not cases:
        raise InvalidReport(f'{filename}: no testcases; empty reports are not evidence of a passing job')
    return cases


def read_artifacts(paths):
    if not 1 <= len(paths) <= 100:
        raise InvalidReport('Use between 1 and 100 XML files per attempt')
    artifacts, cases, names, keys = [], [], set(), set()
    total = 0
    for given in paths:
        path = Path(given)
        if path.suffix.lower() != '.xml' or path.is_symlink() or not path.is_file():
            raise InvalidReport(f'{path}: expected a regular .xml file, not a symbolic link')
        name = path.name
        identity(name, 'artifact filename')
        if name in names:
            raise InvalidReport(f'Duplicate artifact basename: {name}')
        names.add(name)
        with path.open('rb') as handle:
            content = handle.read(MAX_BYTES + 1)
        total += len(content)
        if total > MAX_BYTES:
            raise InvalidReport('Combined attempt artifact size exceeds 10 MiB')
        parsed = parse_xml(content, name)
        for case in parsed:
            if case['key'] in keys:
                raise InvalidReport(f'Duplicate test identity across artifacts: {case["key"]}')
            keys.add(case['key'])
        cases.extend(parsed)
        if len(cases) > MAX_CASES:
            raise InvalidReport('Combined attempt exceeds testcase limit')
        artifacts.append(dict(name=name, content=content, sha256=hashlib.sha256(content).hexdigest()))
    artifacts.sort(key=lambda x: x['name'])
    fingerprint = hashlib.sha256(json.dumps([(a['name'], a['sha256']) for a in artifacts], separators=(',', ':')).encode()).hexdigest()
    return artifacts, cases, fingerprint


SCHEMA = '''
CREATE TABLE IF NOT EXISTS builds(name TEXT PRIMARY KEY, sequence INTEGER NOT NULL UNIQUE);
CREATE TABLE IF NOT EXISTS attempts(
 build TEXT NOT NULL REFERENCES builds(name), job TEXT NOT NULL, attempt INTEGER NOT NULL,
 mode TEXT NOT NULL CHECK(mode IN ('full','partial')), fingerprint TEXT NOT NULL,
 PRIMARY KEY(build,job,attempt));
CREATE TABLE IF NOT EXISTS artifacts(
 build TEXT NOT NULL, job TEXT NOT NULL, attempt INTEGER NOT NULL, name TEXT NOT NULL,
 sha256 TEXT NOT NULL, content BLOB NOT NULL,
 PRIMARY KEY(build,job,attempt,name),
 FOREIGN KEY(build,job,attempt) REFERENCES attempts(build,job,attempt));
CREATE TABLE IF NOT EXISTS cases(
 build TEXT NOT NULL, job TEXT NOT NULL, attempt INTEGER NOT NULL, test_key TEXT NOT NULL,
 file TEXT NOT NULL, classname TEXT NOT NULL, name TEXT NOT NULL,
 status TEXT NOT NULL CHECK(status IN ('pass','failure','error','skipped')),
 duration REAL, message TEXT NOT NULL, message_truncated INTEGER NOT NULL,
 artifact TEXT NOT NULL, PRIMARY KEY(build,job,attempt,test_key),
 FOREIGN KEY(build,job,attempt,artifact) REFERENCES artifacts(build,job,attempt,name));
CREATE INDEX IF NOT EXISTS case_identity ON cases(job,test_key);
'''


def connect(path, create=True):
    path = Path(path)
    if create:
        path.parent.mkdir(parents=True, exist_ok=True)
        db = sqlite3.connect(path, timeout=3, isolation_level=None)
        app_id = db.execute('PRAGMA application_id').fetchone()[0]
        version = db.execute('PRAGMA user_version').fetchone()[0]
        tables = db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
        supported = app_id == APPLICATION_ID and version == 1
        empty = app_id == 0 and version == 0 and not tables
        if not supported and not empty:
            db.close()
            raise InvalidReport('Database is unrelated or has an unsupported schema version')
        db.execute('PRAGMA foreign_keys=ON')
        if empty:
            # Initialize only a genuinely empty database. Existing supported
            # databases need no DDL or header writes when a connection opens.
            try:
                db.executescript('BEGIN IMMEDIATE;\n' + SCHEMA +
                                 f'PRAGMA application_id={APPLICATION_ID};\n'
                                 'PRAGMA user_version=1;\nCOMMIT;')
            except Exception:
                if db.in_transaction:
                    db.execute('ROLLBACK')
                db.close()
                raise
    else:
        if not path.is_file():
            raise InvalidReport(f'Database does not exist: {path}')
        db = sqlite3.connect(path.resolve().as_uri() + '?mode=ro', uri=True, timeout=3, isolation_level=None)
        if db.execute('PRAGMA user_version').fetchone()[0] != 1 or db.execute('PRAGMA application_id').fetchone()[0] != APPLICATION_ID:
            db.close()
            raise InvalidReport('Unsupported database schema version')
    db.row_factory = sqlite3.Row
    return db


def ingest(db_path, paths, *, build, sequence, job, attempt=1, mode='full', dry_run=False, fault_after=None):
    identity(build, 'build'); identity(job, 'job')
    positive_int(sequence, 'sequence'); positive_int(attempt, 'attempt')
    if mode not in ('full', 'partial'):
        raise InvalidReport('mode must be full or partial')
    artifacts, cases, fingerprint = read_artifacts(paths)
    # Parsing and dry-run validation do not create or alter a database.
    if dry_run:
        return {'validated': True, 'cases': len(cases), 'artifacts': len(artifacts), 'fingerprint': fingerprint,
                'note': 'XML validation only; database identity and retry prerequisites are checked by ingest.'}
    db = connect(db_path)
    try:
        db.execute('BEGIN IMMEDIATE')
        prior = db.execute('SELECT * FROM builds WHERE name=? OR sequence=?', (build, sequence)).fetchall()
        if prior and (len(prior) != 1 or prior[0]['name'] != build or prior[0]['sequence'] != sequence):
            raise Conflict('Build identity or sequence is already assigned differently')
        existing = db.execute('SELECT * FROM attempts WHERE build=? AND job=? AND attempt=?', (build,job,attempt)).fetchone()
        if existing:
            if existing['fingerprint'] != fingerprint or existing['mode'] != mode:
                raise Conflict('Attempt identity is immutable; conflicting artifact or mode')
            db.execute('ROLLBACK')
            return {'ingested': False, 'duplicate': True, 'cases': len(cases), 'fingerprint': fingerprint}
        if mode == 'partial' and not db.execute('SELECT 1 FROM attempts WHERE build=? AND job=? AND attempt<? AND mode=\'full\'', (build,job,attempt)).fetchone():
            raise InvalidReport('A partial retry needs an earlier full attempt for this build/job')
        db.execute('INSERT OR IGNORE INTO builds VALUES(?,?)', (build,sequence))
        db.execute('INSERT INTO attempts VALUES(?,?,?,?,?)', (build,job,attempt,mode,fingerprint))
        for artifact in artifacts:
            db.execute('INSERT INTO artifacts VALUES(?,?,?,?,?,?)', (build,job,attempt,artifact['name'],artifact['sha256'],artifact['content']))
        for index, case in enumerate(cases,1):
            db.execute('INSERT INTO cases VALUES(?,?,?,?,?,?,?,?,?,?,?,?)',
                       (build,job,attempt,case['key'],case['file'],case['classname'],case['name'],case['status'],case['duration'],case['message'],int(case['message_truncated']),case['artifact']))
            if fault_after is not None and index == fault_after:
                raise RuntimeError('Injected interruption before commit')
        db.execute('COMMIT')
        return {'ingested': True, 'duplicate': False, 'cases': len(cases), 'fingerprint': fingerprint}
    except Exception:
        if db.in_transaction:
            db.execute('ROLLBACK')
        raise
    finally:
        db.close()


def snapshot(db_path):
    db = connect(db_path, False)
    try:
        db.execute('BEGIN')
        builds = [dict(row) for row in db.execute('SELECT * FROM builds ORDER BY sequence')]
        attempts = [dict(row) for row in db.execute('SELECT * FROM attempts ORDER BY build,job,attempt')]
        cases = [dict(row) for row in db.execute('SELECT * FROM cases ORDER BY build,job,attempt,test_key')]
        artifacts = [{**dict(row), 'content': base64.b64encode(row['content']).decode('ascii')} for row in db.execute('SELECT * FROM artifacts ORDER BY build,job,attempt,name')]
        db.execute('COMMIT')
    finally:
        db.close()
    grouped = {}
    for case in cases:
        grouped.setdefault((case['build'],case['job'],case['attempt']),[]).append(case)
    final, chains, seen = {}, {}, {}
    for build in builds:
        state, chain = {}, {}
        for attempt in [a for a in attempts if a['build'] == build['name']]:
            job = attempt['job']
            if attempt['mode'] == 'full':
                state = {key: value for key,value in state.items() if key[0] != job}
            for case in grouped[(build['name'],job,attempt['attempt'])]:
                key = (job,case['test_key'])
                state[key] = case
                chain.setdefault(key,[]).append(case)
                seen.setdefault(key,dict(job=job,key=case['test_key'],file=case['file'],classname=case['classname'],name=case['name'],first_sequence=build['sequence']))
        final[build['name']] = state
        chains[build['name']] = chain
    selections = {}
    for index, build in enumerate(builds):
        name = build['name']; prior = final[builds[index-1]['name']] if index else {}
        rows = []
        for key, test in sorted(seen.items(), key=lambda item:(item[0][0],item[1]['classname'],item[1]['name'],item[1]['file'])):
            if test['first_sequence'] > build['sequence']:
                continue
            case = final[name].get(key)
            previous = prior.get(key)
            chain = chains[name].get(key,[])
            status = case['status'] if case else 'not reported'
            if not case or status == 'skipped':
                category = status
            elif status in FAILED:
                category = 'repeated failure' if previous and previous['status'] in FAILED else 'new failure' if previous and previous['status']=='pass' else 'failure without baseline'
            elif any(c['status'] in FAILED for c in chain[:-1]):
                category = 'retry pass'
            elif previous and previous['status'] in FAILED:
                category = 'recovered'
            else:
                category = 'pass'
            history=[]
            for earlier in builds[:index+1]:
                value=final[earlier['name']].get(key)
                trail=chains[earlier['name']].get(key,[])
                history.append({'build':earlier['name'],'status':value['status'] if value else 'not reported',
                                'retry_pass':bool(value and value['status']=='pass' and any(c['status'] in FAILED for c in trail[:-1])),
                                'duration':value['duration'] if value else None})
            rows.append({**test,'status':status,'category':category,'duration':case['duration'] if case else None,
                         'message':case['message'] if case else 'No result in the latest full attempt or a later partial retry.',
                         'attempt':case['attempt'] if case else None,'artifact':case['artifact'] if case else None,
                         'history':history,'attempts':[{'attempt':c['attempt'],'status':c['status'],'duration':c['duration'],'message':c['message'],'artifact':c['artifact']} for c in chain]})
        counts={status:sum(row['category']==status for row in rows) for status in STATUSES}
        selections[name]={'rows':rows,'counts':counts,'failed':sum(row['status'] in FAILED for row in rows),
                          'execution_seconds':round(sum(c['duration'] or 0 for c in cases if c['build']==name),6),
                          'attempts':sum(a['build']==name for a in attempts),'observations':sum(c['build']==name for c in cases)}
    return {'schema_version':1,'builds':builds,'selections':selections,'artifacts':artifacts,
            'policy':'One repository/comparison stream per database. Explicit sequence defines build order. Full snapshots replace job membership; partial retries overlay observed cases. Absence is not a pass. Retry pass is a candidate for investigation, not a causal diagnosis.'}


def csv_text(selection):
    output=io.StringIO(newline='');writer=csv.writer(output)
    writer.writerow(['job','file','classname','name','category','status','duration_seconds','attempt'])
    def safe(value):
        text='' if value is None else str(value)
        return "'"+text if text and (text[0] in '=+-@\t\r\n' or text.lstrip().startswith(('=','+','-','@'))) else text
    for row in selection['rows']:
        writer.writerow([safe(row[k]) for k in ['job','file','classname','name','category','status','duration','attempt']])
    return output.getvalue()


def atomic_write(path, content):
    path=Path(path);path.parent.mkdir(parents=True,exist_ok=True)
    fd,temp=tempfile.mkstemp(prefix='.'+path.name+'-',dir=path.parent)
    try:
        with os.fdopen(fd,'wb') as handle:
            handle.write(content.encode('utf-8') if isinstance(content,str) else content);handle.flush();os.fsync(handle.fileno())
        os.replace(temp,path)
    finally:
        if os.path.exists(temp):os.unlink(temp)


def export(db_path,out_dir):
    from report import render
    data=snapshot(db_path)
    if not data['builds']:
        raise InvalidReport('No builds have been ingested')
    out=Path(out_dir)
    atomic_write(out/'analysis.json',json.dumps(data,indent=2,ensure_ascii=True)+'\n')
    latest=data['builds'][-1]['name']
    atomic_write(out/'latest.csv',csv_text(data['selections'][latest]))
    atomic_write(out/'index.html',render(data))
    return data


def main(argv=None):
    parser=argparse.ArgumentParser(description=__doc__)
    commands=parser.add_subparsers(dest='command',required=True)
    add=commands.add_parser('ingest',help='Validate and ingest one immutable job attempt')
    add.add_argument('xml',nargs='+');add.add_argument('--db',default='triage.sqlite')
    add.add_argument('--build',required=True);add.add_argument('--sequence',required=True,type=int)
    add.add_argument('--job',required=True);add.add_argument('--attempt',type=int,default=1)
    add.add_argument('--mode',choices=['full','partial'],default='full');add.add_argument('--dry-run',action='store_true')
    report=commands.add_parser('report',help='Export standalone HTML, JSON and CSV')
    report.add_argument('--db',default='triage.sqlite');report.add_argument('--out',default='triage-report')
    check=commands.add_parser('check',help='Return 1 for final failures in selected build, 0 otherwise; 2 invalid input')
    check.add_argument('--db',default='triage.sqlite');check.add_argument('--build')
    args=parser.parse_args(argv)
    try:
        if args.command=='ingest':
            result=ingest(args.db,args.xml,build=args.build,sequence=args.sequence,job=args.job,attempt=args.attempt,mode=args.mode,dry_run=args.dry_run)
        elif args.command=='report':
            data=export(args.db,args.out);result={'builds':len(data['builds']),'output':str(Path(args.out)/'index.html')}
        else:
            data=snapshot(args.db)
            if not data['builds']:raise InvalidReport('No builds have been ingested')
            name=args.build or data['builds'][-1]['name']
            if name not in data['selections']:raise InvalidReport('Unknown build')
            result={'build':name,'failed':data['selections'][name]['failed'],'categories':data['selections'][name]['counts']}
            print(json.dumps(result,sort_keys=True));return 1 if result['failed'] else 0
        print(json.dumps(result,sort_keys=True));return 0
    except (InvalidReport,Conflict,sqlite3.Error,OSError) as exc:
        print(json.dumps({'error':str(exc)}),file=sys.stderr);return 2

if __name__=='__main__':
    raise SystemExit(main())