intelligence-radar/intelligence_radar/library_briefs.py

"""Retain automation briefs that inbound-library routed to this repository.

The library is reached only through its pull CLI (`new`, `show`, `get`). Grok
mails a notification rather than the brief: the automation name, the title, a
teaser cut off mid-sentence, and a "Continue reading" link that needs a
signed-in session. Such a mail is retained as a `partial` brief, linked to its
period. The full text still arrives through the drop folder (external_briefs).
No session is sent to grok.com.
"""
import hashlib
import json
import re
import subprocess
import sys
from datetime import date, datetime
from pathlib import Path
from zoneinfo import ZoneInfo

from .ingest import MAX_BYTES
from .store import all_records, atomic, put

AUTOMATION = re.compile(r'intelligence-radar-(daily|weekly)', re.IGNORECASE)
CONTINUE = re.compile(r'Continue reading\s*<?(https://grok\.com/\S+?)>?\s', re.IGNORECASE)
FORWARD_MARKER = re.compile(r'-{3,}\s*(?:Forwarded message|Weitergeleitete Nachricht)\s*-{3,}',
                            re.IGNORECASE)
FORWARD_DATE = re.compile(r'^\s*(?:Date|Datum)\s*:\s*(.+)$', re.MULTILINE)
MONTHS = {'jan': 1, 'feb': 2, 'mar': 3, 'mär': 3, 'maer': 3, 'apr': 4, 'may': 5, 'mai': 5,
          'jun': 6, 'jul': 7, 'aug': 8, 'sep': 9, 'oct': 10, 'okt': 10, 'nov': 11,
          'dec': 12, 'dez': 12}


class LibraryUnavailable(Exception):
    pass


def library_cli(spec, *args):
    """Run one inbound-library pull command; return its stdout bytes."""
    root = Path(spec['root']).expanduser()
    if not (root / 'inbound_library').is_dir():
        raise LibraryUnavailable(f'inbound-library not found at {root}')
    done = subprocess.run([sys.executable, '-m', 'inbound_library', '--root', str(root), *args],
                          cwd=root, capture_output=True, timeout=60)
    if done.returncode:
        raise LibraryUnavailable(f"inbound-library {args[0]} exited {done.returncode}")
    return done.stdout


def original_date(text):
    """Send date of the forwarded original ("So., 4. Okt. 2026 um 20:07 Uhr" or
    "Sun, Oct 4, 2026 at 8:07 PM"), or None."""
    marker = FORWARD_MARKER.search(text or '')
    match = FORWARD_DATE.search(text, marker.end()) if marker else None
    if not match:
        return None
    value = match.group(1)
    german = re.search(r'(\d{1,2})\.\s*([A-Za-zäÄ]+)\.?\s+(\d{4})', value)
    english = re.search(r'([A-Za-z]{3})[a-z]*\.?\s+(\d{1,2}),?\s+(\d{4})', value)
    try:
        if german:
            day, month, year = german.groups()
            return date(int(year), MONTHS[month[:3].lower()], int(day))
        if english:
            month, day, year = english.groups()
            return date(int(year), MONTHS[month[:3].lower()], int(day))
    except (KeyError, ValueError):
        return None
    return None


def classify(text, received_at, zone='Europe/Berlin'):
    """Cadence and period of a mailed brief. Returns None when it is not one."""
    automation = AUTOMATION.search(text or '')
    if not automation:
        return None
    cadence = automation.group(1).lower()
    sent = original_date(text)
    basis = 'forwarded_date'
    if sent is None:
        sent = datetime.fromisoformat(received_at).astimezone(ZoneInfo(zone)).date()
        basis = 'received_at'
    if cadence == 'weekly':
        iso = sent.isocalendar()
        period = f'{iso.year}-W{iso.week:02}'
    else:
        period = sent.isoformat()
    continued = CONTINUE.search(text + '\n')
    return {'cadence': cadence, 'period': period, 'filename_date': sent.isoformat(),
            'period_basis': basis, 'completeness': 'partial' if continued else 'complete',
            'continue_url': continued.group(1) if continued else ''}


def pull(root, spec, *, cli=library_cli, now=None, zone='Europe/Berlin'):
    """Retain every routed brief from the configured library sources."""
    root = Path(root)
    moment = now or datetime.now(ZoneInfo(zone))
    try:
        notices = json.loads(cli(spec, 'new', '--repo', spec.get('repo', 'intelligence-radar')))
    except (LibraryUnavailable, OSError, ValueError, subprocess.TimeoutExpired) as exc:
        return {'ok': False, 'status': 'library_unavailable', 'error': str(exc), 'items': []}
    seen = {record['content_sha256'] for _, record in all_records(root, 'external-briefs')} \
        if (root / 'data' / 'external-briefs').is_dir() else set()
    items = []
    for notice in notices:
        provider = spec.get('sources', {}).get(notice['source'])
        if provider is None:
            continue
        item = notice['item']
        if item in seen:
            items.append({'item': item, 'status': 'duplicate'})
            continue
        try:
            items.append(_retain(root, spec, cli, notice, provider, moment.isoformat(), zone))
        except (LibraryUnavailable, OSError, ValueError, subprocess.TimeoutExpired) as exc:
            items.append({'item': item, 'status': 'error', 'error': str(exc)})
            continue
        seen.add(item)
    return {'ok': not any(i['status'] == 'error' for i in items), 'status': 'pulled',
            'items': items}


def _retain(root, spec, cli, notice, provider, imported_at, zone):
    item = notice['item']
    shown = json.loads(cli(spec, 'show', item))
    derived = shown.get('derived') or {}
    text = derived.get('text', '')
    parsed = classify(text, notice['received_at'], zone)
    if parsed is None:
        return {'item': item, 'status': 'not_a_brief'}
    raw = cli(spec, 'get', item)
    if hashlib.sha256(raw).hexdigest() != item:
        raise ValueError(f'library item {item[:12]} does not match its id')
    if len(raw) > MAX_BYTES:
        return {'item': item, 'status': 'too_large'}
    subject = derived.get('subject', '')
    relative = _write_markdown(root, provider, parsed, text, item, imported_at, subject,
                               notice['received_at'])
    atomic(root / 'data' / 'artifacts' / item, raw, immutable=True)
    record = {'schema_version': 2, 'provider': provider, 'cadence': parsed['cadence'],
              'period': parsed['period'], 'filename_date': parsed['filename_date'],
              'source_name': subject, 'content_sha256': item, 'digest_path': relative,
              'imported_at': imported_at, 'origin': 'inbound-library',
              'library_source': notice['source'], 'received_at': notice['received_at'],
              'period_basis': parsed['period_basis'], 'completeness': parsed['completeness'],
              'continue_url': parsed['continue_url']}
    return {'item': item, 'status': 'retained', 'id': put(root, 'external-briefs', record),
            'digest_path': relative, 'cadence': parsed['cadence'], 'period': parsed['period'],
            'completeness': parsed['completeness']}


def _write_markdown(root, provider, parsed, text, item, imported_at, subject, received_at):
    folder = root / 'digests' / 'external' / parsed['cadence']
    folder.mkdir(parents=True, exist_ok=True)
    # The stable <period>-<provider>.md name belongs to the full brief from the drop folder.
    target = folder / f"{parsed['period']}-{provider}-mail-{item[:12]}.md"
    if not target.exists():
        label = ('Partial: mailed teaser, cut off by the provider. The full brief needs the '
                 'drop folder.' if parsed['completeness'] == 'partial' else 'Complete mailed brief.')
        body = '\n'.join([
            '> External automation brief received by email through inbound-library. Unverified. '
            'Not a reviewed digest.', f'> {label}', '',
            f'- Provider: {provider}', f"- Cadence: {parsed['cadence']}",
            f"- Period: {parsed['period']} (from {parsed['period_basis']})",
            f'- Subject: {subject}', f'- Received: {received_at}',
            f'- Library item (SHA-256 of the original message): {item}',
            f"- Continue reading (needs sign-in, not fetched): {parsed['continue_url'] or '-'}",
            f'- Imported: {imported_at}', '', text if text.endswith('\n') else text + '\n'])
        target.write_text(body, encoding='utf-8')
    return target.relative_to(root).as_posix()


def contribute_sources(root, spec):
    """Write this repo's primary sources into the inbound-library registry as shared
    URL pointers. URL and publisher come from catalog/sources.json; topics come from
    spec["contribute"]. Primary capture stays here; the library does not fetch them.

    A file that this repo already wrote is updated in place. A file owned by another
    repo is refused.
    """
    root = Path(root)
    library = Path(spec['root']).expanduser()
    target = library / 'registry' / 'sources'
    if not target.is_dir():
        raise LibraryUnavailable(f'inbound-library registry not found at {target}')
    catalog = {s['id']: s for s in json.loads((root / 'catalog' / 'sources.json').read_text())['sources']}
    owner = spec.get('repo', 'intelligence-radar')
    written = []
    for ident, topics in sorted(spec.get('contribute', {}).items()):
        if ident not in catalog:
            raise ValueError(f'contribute names unknown catalog source {ident}')
        entry = catalog[ident]
        source_id = f'ir-{ident}'
        record = {'id': source_id, 'owner_repo': owner, 'channel': 'url', 'url': entry['url'],
                  'topics': sorted(topics), 'cadence': 'irregular', 'visibility': 'shared',
                  'publisher': entry.get('publisher', ''),
                  'notes': f'Primary source captured by {owner} (catalog/sources.json id '
                           f'{ident}). Pointer only; inbound-library does not fetch it.'}
        path = target / f'{source_id}.json'
        if path.exists() and json.loads(path.read_text()).get('owner_repo') != owner:
            raise ValueError(f'{path.name} belongs to another repo')
        content = json.dumps(record, indent=2, ensure_ascii=False) + '\n'
        status = 'unchanged' if path.exists() and path.read_text() == content else \
            ('updated' if path.exists() else 'created')
        if status != 'unchanged':
            atomic(path, content.encode())
        written.append({'source': source_id, 'status': status})
    return written