"""Fact-based owner pipeline and review queue. No delivery or background jobs.""" import json from datetime import date,datetime,timedelta from zoneinfo import ZoneInfo from fastapi import HTTPException,Request,Query from .store import one,many,now,digest,audit from .workflow_models import SCHEMA,Metadata,WorkflowEdit,Settings,SettingsEdit,Review,calendar_day LONDON=ZoneInfo('Europe/London') CLOSED={'completed','cancelled'} JOB_FIELDS=('id','customer_id','service','message','status','notes','scheduled_date','version') def today():return datetime.now(LONDON).date().isoformat() def local_day(stamp):return datetime.fromisoformat(stamp).astimezone(LONDON).date().isoformat() def fail(message,status=409):raise HTTPException(status,message) def job(db,jid): j=one(db,'SELECT * FROM jobs WHERE id=?',(jid,)) if not j:fail('Job not found.',404) return j def metadata(db,jid): row=one(db,'SELECT content FROM job_workflow WHERE job_id=?',(jid,)) return json.loads(row['content']) if row else Metadata().model_dump() def feedback_basis(db,jid): # Older pinned recovery formats have no workflow table. Their binding is # still the original job version; migration never revives old permission. if db.execute("SELECT 1 FROM sqlite_schema WHERE name='job_workflow' AND type='table'").fetchone(): row=one(db,'SELECT feedback_version FROM job_workflow WHERE job_id=?',(jid,)) if row:return row['feedback_version'] return job(db,jid)['version'] def workflow_snapshot(db,jid): j=job(db,jid);w=one(db,'SELECT * FROM job_workflow WHERE job_id=?',(jid,)) return {'job':{k:j[k] for k in JOB_FIELDS},'metadata':json.loads(w['content']),'feedback_version':w['feedback_version'],'version':w['version']} def record_job(db,jid,actor): v=workflow_snapshot(db,jid) db.execute('INSERT INTO job_workflow_history(job_id,version,job_version,actor,snapshot,created_at) VALUES(?,?,?,?,?,?)',(jid,v['version'],v['job']['version'],actor,json.dumps(v,ensure_ascii=False),now())) def register_job(db,jid): from .customer_review import exists,register if exists(db):register(db,job(db,jid)['customer_id']) if not one(db,'SELECT job_id FROM job_workflow WHERE job_id=?',(jid,)): db.execute('INSERT INTO job_workflow VALUES(?,?,?,1)',(jid,json.dumps(Metadata().model_dump()),job(db,jid)['version'])) record_job(db,jid,'initialisation') def setup(db): db.executescript(SCHEMA) for row in many(db,'SELECT id FROM jobs'):register_job(db,row['id']) db.execute('INSERT OR IGNORE INTO opportunity_settings VALUES(1,?,1)',(json.dumps(Settings().model_dump()),)) def enrich(db,rows): return [{**r,'workflow':metadata(db,r['id'])} for r in rows] def edit_job(db,jid,payload,actor): old=job(db,jid) if old['version']!=payload.version:fail('This job changed. Reload and review before saving.') if payload.status=='scheduled' and not payload.scheduled_date:fail('Choose a target date for a scheduled job.',422) register_job(db,jid);before=metadata(db,jid) after=payload.metadata.model_dump() if isinstance(payload,WorkflowEdit) else dict(before) if after['last_contact_date']>today() or after['completed_date']>today():fail('Recorded contact and completion dates cannot be in the future.',422) if payload.status=='completed' and old['status']!='completed' and not after['completed_date']:after['completed_date']=today() if payload.status!='completed':after['completed_date']='' if after['archived'] and payload.status not in CLOSED:fail('Complete or cancel the job before archiving it; unarchive it before reopening.') if after['archived'] and not before['archived']: guards=[("SELECT 1 FROM field_sessions WHERE job_id=? AND state IN ('running','paused','interrupted','stopped','reviewed')",(jid,)), ("SELECT 1 FROM recurring_contracts WHERE job_id=? AND state IN ('active','paused')",(jid,)), ('SELECT 1 FROM calendar_bookings WHERE job_id=? AND archived=0 AND end_date>=?',(jid,today())), ("SELECT 1 FROM document_sources WHERE job_id=? AND state='reserved'",(jid,)), ("SELECT 1 FROM invoices WHERE job_id=? AND status='draft'",(jid,))] if any(db.execute(sql,args).fetchone() for sql,args in guards):fail('Review open capture, repeat work, future bookings or draft billing before archiving this job.') material=old['status']!=payload.status or old['scheduled_date']!=payload.scheduled_date or any(before[k]!=after[k] for k in ('completed_date','archived')) db.execute('UPDATE jobs SET status=?,notes=?,scheduled_date=?,version=version+1 WHERE id=?',(payload.status,payload.notes,payload.scheduled_date,jid)) db.execute('UPDATE job_workflow SET content=?,feedback_version=CASE WHEN ? THEN ? ELSE feedback_version END,version=version+1 WHERE job_id=?',(json.dumps(after),material,old['version']+1,jid)) if material: from .feedback import invalidate invalidate(db,job_id=jid,actor=actor,detail='Job stage, work date, completion date or archive status changed; review feedback permission again.') record_job(db,jid,actor);audit(db,'job.workflow_updated',jid) return {'ok':True,'id':jid,'version':old['version']+1} def settings(db): row=one(db,'SELECT * FROM opportunity_settings WHERE id=1') return {**json.loads(row['content']),'version':row['version']} def pipeline(db,*,q='',stage='all',priority='all',tag='',archive='active',due='all'): rows=enrich(db,many(db,'SELECT j.*,c.name,c.email,c.phone,c.address,c.version AS customer_version FROM jobs j JOIN customers c ON c.id=j.customer_id ORDER BY j.created_at DESC,j.rowid DESC')) total=len(rows);needle=q.casefold().strip();current=today() # Keep each filter independent; the due filter cannot bypass search/status. def selected(j): m=j['workflow'];text=' '.join(str(j[k]) for k in ('id','customer_id','name','email','phone','address','service','message','notes'))+' '+m['next_action']+' '+' '.join(m['tags']) due_ok=due=='all' or bool(m['next_action_date']) and (m['next_action_date']=cfg['quote_days']:add('quote',j,q['id'],f"Current shared quote; {age} days since sharing or the last recorded contact ({basis}).",quote=q,age=age) groups={cid:[j for j in jobs if j['customer_id']==cid] for cid in {j['customer_id'] for j in jobs}} repeat=[] for cid,rows in groups.items(): if cid in active_recurrence or any(j['status'] not in CLOSED and not j['workflow']['archived'] for j in rows):continue complete=[j for j in rows if j['status']=='completed' and j['workflow']['completed_date']] if not complete:continue j=max(complete,key=lambda j:(j['workflow']['completed_date'],j['created_at'],j['id'])) if excluded(j):continue basis=max(j['workflow']['completed_date'],max(x['workflow']['last_contact_date'] for x in rows));age=(date_today-date.fromisoformat(basis)).days if age>=cfg['repeat_days']: repeat.append(j);add('repeat',j,j['id']+':'+j['workflow']['completed_date'],f"Latest recorded completed work was {j['workflow']['completed_date']}; {age} days since work or recorded contact. No open job or active/paused repeat contract.",age=age) for e in eligibility_rows(db): j=by_id[e['id']] if e['eligible'] and not e['previous'] and not excluded(j):add('review',j,j['id'],'Completed exact job with no prior feedback request. All ratings are welcome; public permission is separate.') if cfg['nearby_enabled']: for j in repeat: group=j['workflow']['postcode_group'] if not group:continue visits=[v for v in jobs if not excluded(v) and v['id']!=j['id'] and v['customer_id']!=j['customer_id'] and v['status']=='scheduled' and v['workflow']['postcode_group']==group and day<=v['scheduled_date']<=(date_today+timedelta(days=cfg['nearby_horizon_days'])).isoformat()] if visits: visit=min(visits,key=lambda x:(x['scheduled_date'],x['id']));add('postcode',j,visit['id']+':'+visit['scheduled_date'],f"Same explicitly saved postcode group {group} as planned job {visit['id'][:8]} on {visit['scheduled_date']}. Group match only; no distance or map lookup.",related=visit) return result def queue(db): current=candidates(db);saved={r['id']:r for r in many(db,'SELECT * FROM opportunity_reviews')};output=[] for key in sorted(set(current)|set(saved)): row=current.get(key);r=saved.get(key) if row is None:row={**json.loads(r['context']),'eligible':False,'context_hash':None,'suggested_draft':''} status=r['state'] if r else 'open' effective='open' if status=='snoozed' and r['snooze_until']<=today() else status output.append({**row,'state':status,'effective_state':effective,'version':r['version'] if r else 0,'draft':r['draft'] if r else '', 'note':r['note'] if r else '', 'snooze_until':r['snooze_until'] if r else '', 'delivery':'Not confirmed. Drafts and manual action records do not send messages.'}) return sorted(output,key=lambda r:(r['due_date'] or '9999',r['kind'],r['id'])) def mount_workflow(app,store,session,once): @app.get('/api/pipeline') def listing(request:Request,q:str=Query('',max_length=200),stage:str=Query('all',pattern='^(all|new|site_visit|accepted|scheduled|in_progress|completed|on_hold|cancelled)$'),priority:str=Query('all',pattern='^(all|low|normal|high|urgent)$'),tag:str=Query('',max_length=40),archive:str=Query('active',pattern='^(active|archived|all)$'),due:str=Query('all',pattern='^(all|due|overdue)$')): s=session(request) with store.tenant(s['business_id']) as db:return pipeline(db,q=q,stage=stage,priority=priority,tag=tag,archive=archive,due=due) @app.put('/api/jobs/{jid}/workflow') def update(jid:str,payload:WorkflowEdit,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db:return once(db,request,payload.model_dump(),lambda:edit_job(db,jid,payload,s['subject'])) @app.get('/api/jobs/{jid}/workflow-history') def history(jid:str,request:Request): s=session(request) with store.tenant(s['business_id']) as db: job(db,jid);return [{**r,'snapshot':json.loads(r['snapshot'])} for r in many(db,'SELECT version,job_version,actor,snapshot,created_at FROM job_workflow_history WHERE job_id=? ORDER BY version',(jid,))] @app.get('/api/opportunity-settings') def get_settings(request:Request): s=session(request) with store.tenant(s['business_id']) as db:return settings(db) @app.put('/api/opportunity-settings') def put_settings(payload:SettingsEdit,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db: def act(): if settings(db)['version']!=payload.version:fail('Settings changed. Reload before saving.') content=json.dumps(payload.model_dump(exclude={'version'}));version=payload.version+1 db.execute('UPDATE opportunity_settings SET content=?,version=? WHERE id=1',(content,version)) db.execute('INSERT INTO opportunity_settings_history VALUES(?,?,?,?)',(version,content,s['subject'],now()));audit(db,'opportunities.settings_updated','1');return {'version':version} return once(db,request,payload.model_dump(),act) @app.get('/api/opportunities') def opportunities(request:Request): s=session(request) with store.tenant(s['business_id']) as db: rows=queue(db);return {'items':rows,'today':today(),'ready':sum(r['eligible'] and r['effective_state'] in ('open','prepared') for r in rows),'settings':settings(db)} @app.get('/api/opportunities/{oid}') def detail(oid:str,request:Request): s=session(request) with store.tenant(s['business_id']) as db: row=next((r for r in queue(db) if r['id']==oid),None) if not row:fail('Opportunity not found.',404) return {'item':row,'history':[{**r,'snapshot':json.loads(r['snapshot'])} for r in many(db,'SELECT version,action,actor,snapshot,created_at FROM opportunity_history WHERE opportunity_id=? ORDER BY version',(oid,))]} @app.post('/api/opportunities/{oid}/actions') def action(oid:str,payload:Review,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db: def act(): row=candidates(db).get(oid) if not row: if one(db,'SELECT id FROM opportunity_reviews WHERE id=?',(oid,)):fail('This item is no longer eligible. Review its source records.') fail('Opportunity not found.',404) if row['context_hash']!=payload.context_hash:fail('The source facts changed. Reload and review this item.') prior=one(db,'SELECT * FROM opportunity_reviews WHERE id=?',(oid,));version=prior['version'] if prior else 0 if version!=payload.version:fail('This review changed. Reload before continuing.') if prior and prior['state'] in ('dismissed','completed') and payload.action!='reopen':fail('Reopen this review before taking another action.') if payload.action=='reopen' and (not prior or prior['state'] not in ('dismissed','completed','snoozed')):fail('Only a closed or snoozed review can be reopened.') if payload.action=='snooze' and not today()