tegwick
1d4c55ef25
IR-WP-0010-T03: pull-briefs and a run-due step read routed items through the library CLI, keep the original message as the artifact, classify cadence and period, and record schema-2 briefs marked partial with the unfetched Continue-reading link. Retains the two 4 October Grok mails. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 7780@bnt-lap001 Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
187 lines
8 KiB
Python
187 lines
8 KiB
Python
"""Retain ChatGPT and Grok automation briefs dropped in a local inbox.
|
|
|
|
The inbox's archive directory is already captured and is ignored. Retained text
|
|
is unverified automation output, not a reviewed digest. Website automations on
|
|
chatgpt.com and grok.com do not publish a result feed; this module does not
|
|
send an account session to either site.
|
|
"""
|
|
import hashlib
|
|
import os
|
|
import re
|
|
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
|
|
|
|
NAME = re.compile(
|
|
r'^(?P<ymd>\d{6})-intelligenceRadar(?P<cadence>Daily|Weekly)_(?P<provider>chatgpt|grok)\.txt$')
|
|
RECORD_FIELDS = {'schema_version', 'provider', 'cadence', 'period', 'filename_date',
|
|
'source_name', 'content_sha256', 'digest_path', 'imported_at'}
|
|
# Schema 2: a brief retained from an inbound-library mail (library_briefs.py).
|
|
RECORD_FIELDS_V2 = RECORD_FIELDS | {'origin', 'library_source', 'received_at', 'period_basis',
|
|
'completeness', 'continue_url'}
|
|
SCHEMAS = {1: RECORD_FIELDS, 2: RECORD_FIELDS_V2}
|
|
|
|
NATIVE_STATUS = (
|
|
{'provider': 'chatgpt', 'status': 'needs_session',
|
|
'detail': 'ChatGPT scheduled-task output stays in the signed-in account. '
|
|
'A public share is one conversation snapshot, not a feed of later runs. '
|
|
'Drop the exported text in the brief inbox.'},
|
|
{'provider': 'grok', 'status': 'needs_session',
|
|
'detail': 'grok.com automations keep each run in the signed-in account. '
|
|
'There is no public link or export API. Email delivery can save a run; '
|
|
'place that file in the brief inbox. No account session is sent.'},
|
|
)
|
|
|
|
|
|
def resolve_inbox(spec, environ=None):
|
|
env = os.environ if environ is None else environ
|
|
override = env.get('INTELLIGENCE_RADAR_BRIEF_INBOX')
|
|
if override:
|
|
return Path(override)
|
|
if not spec.get('inbox'):
|
|
raise ValueError('external brief inbox is not configured')
|
|
return Path(spec['inbox']).expanduser()
|
|
|
|
|
|
def parse_name(name):
|
|
match = NAME.fullmatch(name)
|
|
if not match:
|
|
return None
|
|
ymd = match.group('ymd')
|
|
try:
|
|
event = date(2000 + int(ymd[:2]), int(ymd[2:4]), int(ymd[4:6]))
|
|
except ValueError:
|
|
return None
|
|
cadence = match.group('cadence').lower()
|
|
if cadence == 'weekly':
|
|
iso = event.isocalendar()
|
|
period = f'{iso.year}-W{iso.week:02}'
|
|
else:
|
|
period = event.isoformat()
|
|
return {'filename_date': event.isoformat(), 'cadence': cadence,
|
|
'provider': match.group('provider'), 'period': period}
|
|
|
|
|
|
def native_status():
|
|
return {'ok': True, 'status': 'needs_session', 'providers': list(NATIVE_STATUS)}
|
|
|
|
|
|
def briefs_for_period(root, period):
|
|
directory = Path(root) / 'data' / 'external-briefs'
|
|
if not directory.is_dir():
|
|
return []
|
|
return [record for _, record in all_records(root, 'external-briefs')
|
|
if record['period'] == period]
|
|
|
|
|
|
def verify(root):
|
|
root = Path(root)
|
|
directory = root / 'data' / 'external-briefs'
|
|
if not directory.exists():
|
|
return 0
|
|
external = (root / 'digests' / 'external').resolve()
|
|
count = 0
|
|
for ident, record in all_records(root, 'external-briefs'):
|
|
if set(record) != SCHEMAS.get(record.get('schema_version')):
|
|
raise ValueError(f'invalid external brief: {ident}')
|
|
digest = record['content_sha256']
|
|
blob = (root / 'data' / 'artifacts' / digest).read_bytes()
|
|
if hashlib.sha256(blob).hexdigest() != digest:
|
|
raise ValueError(f'external brief artifact mismatch: {digest}')
|
|
path = (root / record['digest_path']).resolve()
|
|
if not path.is_relative_to(external) or not path.is_file():
|
|
raise ValueError(f'external brief digest escapes retention: {ident}')
|
|
if digest not in path.read_text(encoding='utf-8'):
|
|
raise ValueError(f'external brief digest missing content hash: {ident}')
|
|
count += 1
|
|
return count
|
|
|
|
|
|
def import_inbox(root, inbox, *, archive_dirname='archive', now=None, zone='Europe/Berlin'):
|
|
root = Path(root)
|
|
inbox = Path(inbox)
|
|
if not inbox.is_dir():
|
|
return {'ok': False, 'status': 'inbox_missing', 'inbox': str(inbox), 'files': []}
|
|
archive = inbox / archive_dirname
|
|
if archive.exists() and not archive.is_dir():
|
|
return {'ok': False, 'status': 'archive_not_directory', 'inbox': str(inbox), 'files': []}
|
|
if not archive.resolve().is_relative_to(inbox.resolve()):
|
|
return {'ok': False, 'status': 'archive_escapes_inbox', 'inbox': str(inbox), 'files': []}
|
|
moment = now or datetime.now(ZoneInfo(zone))
|
|
if moment.tzinfo is None:
|
|
raise ValueError('import time requires a timezone')
|
|
seen = {}
|
|
if (root / 'data' / 'external-briefs').is_dir():
|
|
seen = {record['content_sha256']: record
|
|
for _, record in all_records(root, 'external-briefs')}
|
|
files = [_one(root, path, archive, seen, moment.isoformat())
|
|
for path in sorted(inbox.iterdir()) if path.is_file()]
|
|
blocking = {'unreadable', 'too_large'}
|
|
return {'ok': not any(item['status'] in blocking for item in files),
|
|
'status': 'imported', 'inbox': str(inbox), 'files': files}
|
|
|
|
|
|
def _one(root, path, archive, seen, imported_at):
|
|
parsed = parse_name(path.name)
|
|
if parsed is None:
|
|
return {'file': path.name, 'status': 'ignored'}
|
|
data = path.read_bytes()
|
|
if not data.strip():
|
|
return {'file': path.name, 'status': 'empty', **parsed}
|
|
if len(data) > MAX_BYTES:
|
|
return {'file': path.name, 'status': 'too_large', **parsed}
|
|
try:
|
|
text = data.decode('utf-8')
|
|
except UnicodeError:
|
|
return {'file': path.name, 'status': 'unreadable', **parsed}
|
|
digest = hashlib.sha256(data).hexdigest()
|
|
result = {'file': path.name, 'content_sha256': digest, **parsed}
|
|
if digest not in seen:
|
|
relative = _write_markdown(root, parsed, text, digest, imported_at, path.name)
|
|
atomic(root / 'data' / 'artifacts' / digest, data, immutable=True)
|
|
record = {'schema_version': 1, 'provider': parsed['provider'], 'cadence': parsed['cadence'],
|
|
'period': parsed['period'], 'filename_date': parsed['filename_date'],
|
|
'source_name': path.name, 'content_sha256': digest,
|
|
'digest_path': relative, 'imported_at': imported_at}
|
|
result['id'] = put(root, 'external-briefs', record)
|
|
result['digest_path'] = relative
|
|
seen[digest] = record
|
|
result['status'] = 'retained'
|
|
else:
|
|
result['status'] = 'duplicate'
|
|
result['digest_path'] = seen[digest]['digest_path']
|
|
result['archive'] = _archive(path, archive, data)
|
|
return result
|
|
|
|
|
|
def _write_markdown(root, parsed, text, digest, imported_at, source_name):
|
|
folder = root / 'digests' / 'external' / parsed['cadence']
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
stable = folder / f"{parsed['period']}-{parsed['provider']}.md"
|
|
target = stable
|
|
if stable.exists() and digest not in stable.read_text(encoding='utf-8'):
|
|
target = folder / f"{parsed['period']}-{parsed['provider']}-{digest[:12]}.md"
|
|
if not target.exists():
|
|
body = '\n'.join([
|
|
'> External automation brief. Unverified. Not a reviewed digest.', '',
|
|
f"- Provider: {parsed['provider']}", f"- Cadence: {parsed['cadence']}",
|
|
f"- Period: {parsed['period']}", f"- Filename date: {parsed['filename_date']}",
|
|
f"- Source file: {source_name}", f"- Content SHA-256: {digest}",
|
|
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 _archive(path, archive, data):
|
|
archive.mkdir(parents=True, exist_ok=True)
|
|
dest = archive / path.name
|
|
if dest.exists():
|
|
if dest.read_bytes() == data:
|
|
path.unlink()
|
|
return 'removed_duplicate'
|
|
return 'left_in_inbox'
|
|
path.replace(dest)
|
|
return 'moved'
|