@@ -0,0 +1,248 @@
""" Fail-closed conversation identities for local OCR shadow testing. """
from __future__ import annotations
import hashlib
import json
import os
import re
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any , Mapping
SHADOW_CONFIG_ENV = " JWEBOT_CONVERSATION_SHADOW_CONFIG "
SHADOW_CONFIG_VERSION = 1
MAX_SHADOW_TARGETS = 8
MAX_EVENT_TEXT = 5_000
CONVERSATION_KEY = re . compile ( r " ^[a-z0-9][a-z0-9._-] { 0,63}$ " )
TITLE_FINGERPRINT = re . compile ( r " ^[0-9a-f] {64} $ " )
class ConversationConfigError ( ValueError ) :
""" Raised when an opt-in shadow conversation registry is invalid. """
class ConversationKind ( str , Enum ) :
GROUP = " group "
DIRECT = " direct "
def _required_text ( value : object , * , name : str , limit : int ) - > str :
text = str ( value or " " ) . strip ( )
if not text :
raise ConversationConfigError ( f " { name } is required " )
if len ( text ) > limit :
raise ConversationConfigError ( f " { name } exceeds { limit } characters " )
return text
def _trigger_question ( text : str , * , bot_mention : str , direct : bool ) - > str :
mention_prefix = r " (?:[@＠ \ s*)? " if direct else r " [@＠ \ s* "
mention = re . compile (
rf " ^ \ s* { mention_prefix } { re . escape ( bot_mention ) } "
r " (?: \ s*[:： ， \ s*( \ S.*)$ " ,
re . IGNORECASE ,
)
for line in text . splitlines ( ) :
stripped = line . strip ( )
if direct and stripped . startswith ( " / " ) :
return stripped
match = mention . match ( line )
if match :
return match . group ( 1 ) . strip ( )
return " "
@dataclass ( frozen = True )
class ConversationTarget :
""" One explicitly registered title fingerprint in shadow mode. """
key : str
kind : ConversationKind
title_fingerprint : str
sender : str = " "
def __post_init__ ( self ) - > None :
if not CONVERSATION_KEY . fullmatch ( self . key ) :
raise ConversationConfigError ( " conversation key must be an opaque slug " )
if not TITLE_FINGERPRINT . fullmatch ( self . title_fingerprint ) :
raise ConversationConfigError (
" title_fingerprint must be a lowercase SHA-256 digest "
)
if self . kind is ConversationKind . DIRECT :
_required_text ( self . sender , name = " direct sender " , limit = 120 )
elif self . sender :
raise ConversationConfigError ( " group targets cannot define a fixed sender " )
@classmethod
def from_mapping ( cls , value : Mapping [ str , Any ] ) - > " ConversationTarget " :
if not isinstance ( value , Mapping ) :
raise ConversationConfigError ( " conversation target must be an object " )
unknown = set ( value ) - {
" key " ,
" kind " ,
" title_fingerprint " ,
" sender " ,
}
if unknown :
raise ConversationConfigError ( " conversation target has unknown fields " )
try :
kind = ConversationKind ( str ( value . get ( " kind " , " " ) ) )
except ValueError as error :
raise ConversationConfigError ( " conversation kind is invalid " ) from error
return cls (
key = _required_text ( value . get ( " key " ) , name = " conversation key " , limit = 64 ) ,
kind = kind ,
title_fingerprint = _required_text (
value . get ( " title_fingerprint " ) ,
name = " title_fingerprint " ,
limit = 64 ,
) . casefold ( ) ,
sender = str ( value . get ( " sender " , " " ) ) . strip ( ) ,
)
def question ( self , text : str , * , bot_mention : str ) - > str :
return _trigger_question (
text ,
bot_mention = bot_mention ,
direct = self . kind is ConversationKind . DIRECT ,
)
@dataclass ( frozen = True )
class ShadowConversationRegistry :
""" A registry that can observe but can never authorize delivery. """
enabled : bool = False
targets : tuple [ ConversationTarget , . . . ] = ( )
def __post_init__ ( self ) - > None :
keys = [ target . key for target in self . targets ]
fingerprints = [ target . title_fingerprint for target in self . targets ]
if len ( keys ) != len ( set ( keys ) ) :
raise ConversationConfigError ( " conversation keys must be unique " )
if len ( fingerprints ) != len ( set ( fingerprints ) ) :
raise ConversationConfigError ( " title fingerprints must be unique " )
@classmethod
def disabled ( cls ) - > " ShadowConversationRegistry " :
return cls ( )
@classmethod
def from_mapping ( cls , value : Mapping [ str , Any ] ) - > " ShadowConversationRegistry " :
if not isinstance ( value , Mapping ) :
raise ConversationConfigError ( " shadow registry must be an object " )
unknown = set ( value ) - { " format_version " , " enabled " , " targets " }
if unknown :
raise ConversationConfigError ( " shadow registry has unknown fields " )
if value . get ( " format_version " ) != SHADOW_CONFIG_VERSION :
raise ConversationConfigError ( " unsupported shadow registry version " )
enabled = value . get ( " enabled " , False )
if not isinstance ( enabled , bool ) :
raise ConversationConfigError ( " shadow registry enabled must be boolean " )
raw_targets = value . get ( " targets " , [ ] )
if not isinstance ( raw_targets , list ) :
raise ConversationConfigError ( " shadow registry targets must be a list " )
if len ( raw_targets ) > MAX_SHADOW_TARGETS :
raise ConversationConfigError ( " shadow registry has too many targets " )
targets = tuple ( ConversationTarget . from_mapping ( item ) for item in raw_targets )
if any ( target . kind is not ConversationKind . DIRECT for target in targets ) :
raise ConversationConfigError ( " shadow registry accepts direct targets only " )
return cls ( enabled = enabled , targets = targets )
@classmethod
def from_environment (
cls ,
environment : Mapping [ str , str ] | None = None ,
) - > " ShadowConversationRegistry " :
source = environment if environment is not None else os . environ
raw_path = str ( source . get ( SHADOW_CONFIG_ENV , " " ) ) . strip ( )
if not raw_path :
return cls . disabled ( )
path = Path ( raw_path ) . expanduser ( )
if path . is_symlink ( ) or not path . is_file ( ) :
raise ConversationConfigError ( " shadow registry path must be a regular file " )
try :
value = json . loads ( path . read_text ( encoding = " utf-8 " ) )
except ( OSError , json . JSONDecodeError ) as error :
raise ConversationConfigError ( " shadow registry cannot be read " ) from error
if not isinstance ( value , Mapping ) :
raise ConversationConfigError ( " shadow registry must be an object " )
return cls . from_mapping ( value )
def target_for_fingerprint (
self ,
fingerprint : str ,
) - > ConversationTarget | None :
if not self . enabled :
return None
normalized = str ( fingerprint ) . casefold ( )
return next (
(
target
for target in self . targets
if target . title_fingerprint == normalized
) ,
None ,
)
def build_shadow_event (
target : ConversationTarget ,
* ,
observed_at : str ,
text : str ,
body_confidence : float ,
visual_signature : str ,
bot_mention : str ,
observed_sender : str = " " ,
sender_confidence : float = 0.0 ,
ordinal : int = 0 ,
) - > dict [ str , Any ] :
""" Build a non-dispatchable observation for synthetic and shadow tests. """
content = _required_text ( text , name = " event text " , limit = MAX_EVENT_TEXT )
question = target . question ( content , bot_mention = bot_mention )
sender = (
target . sender
if target . kind is ConversationKind . DIRECT
else _required_text ( observed_sender , name = " observed sender " , limit = 120 )
)
effective_sender_confidence = (
1.0
if target . kind is ConversationKind . DIRECT
else max ( 0.0 , min ( 1.0 , float ( sender_confidence ) ) )
)
effective_body_confidence = max ( 0.0 , min ( 1.0 , float ( body_confidence ) ) )
seed = (
f " { target . key } \n { observed_at } \n { ordinal } \n { visual_signature } \n "
f " { sender } \n { content } "
)
return {
" version " : 1 ,
" event_id " : f " shadow- { hashlib . sha256 ( seed . encode ( ) ) . hexdigest ( ) [ : 24 ] } " ,
" observed_at " : observed_at ,
" sender " : sender ,
" text " : content ,
" question " : question ,
" confidence " : min (
effective_sender_confidence ,
effective_body_confidence ,
) ,
" sender_confidence " : effective_sender_confidence ,
" body_confidence " : effective_body_confidence ,
" side " : " incoming " ,
" mention_bot " : bool ( question ) ,
" group_fingerprint " : target . title_fingerprint ,
" conversation_key " : target . key ,
" conversation_kind " : target . kind . value ,
" source " : (
" rapidocr-private-shadow "
if target . kind is ConversationKind . DIRECT
else " rapidocr-group-shadow "
) ,
" shadow_only " : True ,
" dispatch_allowed " : False ,
}
JWeBot Contributors