175 lines
9.9 KiB
Python
175 lines
9.9 KiB
Python
"""Side-effect-free classification primitives shared by simulator and worker."""
|
|
import re
|
|
|
|
|
|
def ancestor_ids(tag_ids, parents):
|
|
result = set()
|
|
for tag_id in tag_ids:
|
|
seen = set()
|
|
current = tag_id
|
|
while current is not None:
|
|
if current in seen:
|
|
raise ValueError('Tag hierarchy contains a cycle')
|
|
seen.add(current)
|
|
result.add(current)
|
|
current = parents.get(current)
|
|
return result
|
|
|
|
|
|
def validate_parent(tag_id, parent_id, parents):
|
|
updated = dict(parents)
|
|
if parent_id is not None and parent_id not in parents:
|
|
raise ValueError('Parent tag does not exist')
|
|
updated[tag_id] = parent_id
|
|
ancestor_ids([tag_id], updated)
|
|
|
|
|
|
def match_keywords(text, rules):
|
|
text = str(text or '').casefold()
|
|
results = []
|
|
for rule in sorted(rules, key=lambda r: (r.get('priority', 100), r.get('id', 0))):
|
|
mode = rule.get('mode', 'contains')
|
|
def matches(word):
|
|
word = str(word).strip().casefold()
|
|
if not word:
|
|
return False
|
|
if mode == 'exact':
|
|
return text.strip() == word
|
|
if mode == 'whole_word':
|
|
return bool(re.search(r'(?<!\w)' + re.escape(word) + r'(?!\w)', text))
|
|
if mode != 'contains':
|
|
raise ValueError('Unknown keyword mode')
|
|
return word in text
|
|
positives = [word for word in rule.get('alternatives', []) if matches(word)]
|
|
if positives and not any(matches(word) for word in rule.get('negative_words', [])):
|
|
results.append({'tag_id': rule['tag_id'], 'rule_id': rule.get('id'),
|
|
'matched_words': positives, 'source': 'keyword'})
|
|
return results
|
|
|
|
|
|
def classify_ai_result(result, tags, automatic_threshold=90, suggestion_threshold=70):
|
|
if not 0 <= suggestion_threshold <= automatic_threshold <= 100:
|
|
raise ValueError('Invalid confidence thresholds')
|
|
tag_id = result.get('tag_id')
|
|
if tag_id not in tags or not tags[tag_id].get('is_active', True):
|
|
raise ValueError('AI returned an invalid tag')
|
|
confidence = float(result.get('confidence', 0))
|
|
if not 0 <= confidence <= 100:
|
|
raise ValueError('Invalid confidence')
|
|
billing = bool(tags[tag_id].get('billing_relevant') or tags[tag_id].get('type') == 'billing')
|
|
status = 'manual_review' if confidence < suggestion_threshold else 'pending'
|
|
if confidence >= automatic_threshold and not billing:
|
|
status = 'automatic'
|
|
return {'tag_id': tag_id, 'confidence': confidence, 'status': status,
|
|
'billing_relevant': billing, 'explanation': str(result.get('explanation') or '')}
|
|
|
|
|
|
def billing_tag_ids(definitions, parents):
|
|
"""Conservatively mark tag references in billing workflows and all descendants."""
|
|
referenced = set()
|
|
def walk(value):
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
if key in ('tag_id', 'tag_ids', 'excluded_tag_ids', 'required_tag_ids'):
|
|
values = child if isinstance(child, list) else [child]
|
|
referenced.update(v for v in values if isinstance(v, int))
|
|
if value.get('field') in ('tags', 'tag_ids', 'tag_id') and key == 'value':
|
|
values = child if isinstance(child, list) else [child]
|
|
referenced.update(v for v in values if isinstance(v, int))
|
|
walk(child)
|
|
elif isinstance(value, list):
|
|
for child in value: walk(child)
|
|
for definition in definitions: walk(definition)
|
|
return {tag for tag in parents if ancestor_ids([tag], parents) & referenced}
|
|
|
|
|
|
def classify_time_on_connection(conn, time_id, text):
|
|
"""Evaluate keyword classification using caller transaction; never remove tags.
|
|
|
|
Returns classification metadata without applying tags. Caller may use matched IDs
|
|
as simulation context. Pending AI proposals block billing regardless of shadow mode.
|
|
"""
|
|
from psycopg2.extras import RealDictCursor
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
|
cur.execute('SELECT id,parent_id FROM tags')
|
|
parents={row['id']:row['parent_id'] for row in cur.fetchall()}
|
|
cur.execute('SELECT k.* FROM workflow_tag_keywords k JOIN tags t ON t.id=k.tag_id WHERE k.active AND t.is_active')
|
|
matches=match_keywords(text,cur.fetchall())
|
|
cur.execute("SELECT tag_id FROM entity_tags WHERE entity_type='time_entry' AND entity_id=%s",(time_id,))
|
|
concrete={row['tag_id'] for row in cur.fetchall()} | {m['tag_id'] for m in matches}
|
|
cur.execute("SELECT id FROM workflow_tag_proposals WHERE entity_type='time_entry' AND entity_id=%s AND billing_relevant AND status IN ('pending','manual_review')",(time_id,))
|
|
pending=[row['id'] for row in cur.fetchall()]
|
|
return {'tag_ids':sorted(concrete),'effective_tag_ids':sorted(ancestor_ids(concrete,parents)),
|
|
'keyword_matches':matches,'pending_billing_proposals':pending}
|
|
|
|
|
|
def enrich_entity_context(cursor, context, entity_type, entity_id):
|
|
"""Shadow-safe keyword and hierarchy enrichment on the worker transaction."""
|
|
cursor.execute('SELECT id,parent_id FROM tags')
|
|
parents={r['id']:r['parent_id'] for r in cursor.fetchall()}
|
|
cursor.execute('SELECT k.* FROM workflow_tag_keywords k JOIN tags t ON t.id=k.tag_id WHERE k.active AND t.is_active')
|
|
matches=match_keywords(classification_input(context),cursor.fetchall())
|
|
if entity_type in ('case', 'time_entry'):
|
|
case_id = entity_id if entity_type == 'case' else context.get('case_id')
|
|
cursor.execute("SELECT tag_id FROM workflow_tag_proposals WHERE source='manual_dismissal' AND status='rejected' AND ((entity_type=%s AND entity_id=%s) OR (entity_type='case' AND entity_id=%s))", (entity_type, entity_id, case_id))
|
|
dismissed = {row['tag_id'] for row in cursor.fetchall()}
|
|
matches = [match for match in matches if match['tag_id'] not in dismissed]
|
|
existing=set(context.get('concrete_tag_ids',context.get('tag_ids',context.get('tags',[]))))
|
|
cursor.execute('SELECT t.id,t.tag_group_id,tg.behavior FROM tags t LEFT JOIN tag_groups tg ON tg.id=t.tag_group_id')
|
|
groups={r['id']:r for r in cursor.fetchall()}
|
|
conflicts=[]
|
|
accepted=[]
|
|
for match in matches:
|
|
group=groups.get(match['tag_id'],{})
|
|
conflict=group.get('behavior') in ('single','toggle') and any(groups.get(tag,{}).get('tag_group_id')==group.get('tag_group_id') and tag!=match['tag_id'] for tag in (existing | {m['tag_id'] for m in matches if m['tag_id']!=match['tag_id']}))
|
|
if conflict: conflicts.append('keyword_group:'+str(match['tag_id']))
|
|
else: accepted.append(match)
|
|
matches=accepted
|
|
concrete=existing | {m['tag_id'] for m in matches}
|
|
effective=sorted(ancestor_ids(concrete,parents))
|
|
cursor.execute("SELECT id FROM workflow_tag_proposals WHERE entity_type=%s AND entity_id=%s AND billing_relevant AND status IN ('pending','manual_review') AND (source<>'ai' OR input_text=%s)",(entity_type,entity_id,classification_input(context)))
|
|
pending=[r['id'] for r in cursor.fetchall()] + conflicts
|
|
if entity_type=='time_entry' and context.get('case_id'):
|
|
case_input=classification_input({'entity_type':'case','subject':context.get('case_title'),'description':context.get('case_description')})
|
|
cursor.execute("SELECT id FROM workflow_tag_proposals WHERE entity_type='case' AND entity_id=%s AND billing_relevant AND status IN ('pending','manual_review') AND (source<>'ai' OR input_text=%s)",(context['case_id'],case_input))
|
|
pending += ['case_proposal:'+str(r['id']) for r in cursor.fetchall()]
|
|
if entity_type=='time_entry':
|
|
cursor.execute("SELECT id FROM workflow_tag_ai_jobs WHERE time_id=%s AND input_hash=%s AND status IN ('pending','processing','failed')",(entity_id,__import__('hashlib').sha256(classification_input(context).encode()).hexdigest()))
|
|
pending += ['ai_job:' + str(r['id']) for r in cursor.fetchall()]
|
|
return dict(context,tags=effective,tag_ids=effective,concrete_tag_ids=sorted(concrete),
|
|
keyword_matches=matches,pending_billing_tag_suggestions=pending)
|
|
|
|
|
|
def classification_input(context):
|
|
if context.get('entity_type') in ('case','mail','email') or (context.get('subject') and not context.get('time_id')):
|
|
return '\n'.join(str(v or '') for v in (context.get('subject') or context.get('case_title'),context.get('description') or context.get('case_description'))).strip()
|
|
return '\n'.join(str(context.get(k) or '') for k in ('description','case_title','case_description')).strip()
|
|
|
|
|
|
def enrich_time_context(cursor,context):
|
|
return enrich_entity_context(cursor,context,'time_entry',context['time_id'])
|
|
|
|
|
|
def enrich_case_context(cursor,context):
|
|
return enrich_entity_context(cursor,context,'case',context.get('case_id') or context['id'])
|
|
|
|
|
|
def enrich_mail_context(cursor,context):
|
|
context=dict(context,entity_type='email',description=context.get('description') or context.get('body_text') or context.get('body_html') or '')
|
|
entity_id=context.get('entity_id') or context['id']
|
|
cursor.execute("SELECT tag_id FROM entity_tags WHERE entity_type='email' AND entity_id=%s",(entity_id,))
|
|
context['concrete_tag_ids']=[r['tag_id'] for r in cursor.fetchall()]
|
|
return enrich_entity_context(cursor,context,'email',entity_id)
|
|
|
|
|
|
def apply_entity_keyword_matches(cursor,entity_type,entity_id,context):
|
|
"""Worker-only persistence; preserve concrete manual assignments and conflicts."""
|
|
if entity_type not in ('time_entry','case','email') or not entity_id: return
|
|
cursor.execute('SELECT pg_advisory_xact_lock(hashtext(%s))',(f"tag:{entity_type}:{entity_id}",))
|
|
for match in context.get('keyword_matches',[]):
|
|
cursor.execute("INSERT INTO entity_tags(entity_type,entity_id,tag_id,source) VALUES(%s,%s,%s,'keyword') ON CONFLICT(entity_type,entity_id,tag_id) DO NOTHING",(entity_type,entity_id,match['tag_id']))
|
|
|
|
|
|
def apply_keyword_matches(cursor,context):
|
|
apply_entity_keyword_matches(cursor,'time_entry',context.get('time_id'),context)
|