"""Owner-private customer history, export and deliberate review endpoints.""" import csv import io import json import unicodedata from fastapi import Query, Request, Response from .store import one,many,now from .documents import invoice_rows from . import customer_review as review from . import customer_review_models as m def history(db,cid): from .workflow import enrich,queue c=review.contact(db,cid) jobs=enrich(db,many(db,'SELECT * FROM jobs WHERE customer_id=? ORDER BY created_at DESC,id',(cid,))) quotes=many(db,'SELECT id,reference,job_id,customer_id,status,total_pence,snapshot,created_at FROM quotes WHERE customer_id=? ORDER BY created_at DESC,id',(cid,)) invoices=invoice_rows(db,'WHERE i.customer_id=?',(cid,)) for d in quotes+invoices: original=json.loads(d.pop('snapshot'))['customer'] d.update(original_customer_name=original['name'],original_customer=original,current_customer_id=cid) payments=many(db,'SELECT p.id,p.invoice_id,p.amount_pence,p.created_at,i.reference,d.paid_on,d.method,CASE WHEN r.payment_id IS NULL THEN 0 ELSE 1 END AS reversed FROM payments p JOIN invoices i ON i.id=p.invoice_id JOIN payment_details d ON d.payment_id=p.id LEFT JOIN payment_reversals r ON r.payment_id=p.id WHERE i.customer_id=? ORDER BY d.paid_on DESC,p.created_at DESC,p.id',(cid,)) links=many(db,'SELECT job_id,from_customer_id,to_customer_id,created_at FROM customer_links WHERE from_customer_id=? OR to_customer_id=? ORDER BY id DESC',(cid,cid)) changes=many(db,'SELECT created_at FROM customer_changes WHERE customer_id=? ORDER BY id DESC',(cid,)) records=[{**r,'snapshot':json.loads(r['snapshot'])} for r in many(db,'SELECT * FROM customer_record_history WHERE customer_id=? ORDER BY version',(cid,))] transfers=[{**r,'plan':json.loads(r['plan']),'result':json.loads(r['result'])} for r in many(db,'SELECT * FROM customer_transfers WHERE source_id=? OR target_id=? ORDER BY rowid',(cid,cid))] historical=[] for kind,table in (('quote','quotes'),('invoice','invoices')): for d in many(db,"SELECT id,reference,job_id,customer_id,snapshot,total_pence,status,created_at FROM "+table+" WHERE json_extract(snapshot,'$.customer.id')=? AND customer_id<>? ORDER BY created_at,id",(cid,cid)): d['original_customer']=json.loads(d.pop('snapshot'))['customer'];d['kind']=kind;historical.append(d) ids={j['id'] for j in jobs} followups=[r for r in queue(db) if r['customer_id']==cid or r['job_id'] in ids] permitted=[] for j in jobs: counts={t:db.execute('SELECT COUNT(*) FROM '+t+' WHERE job_id=?',(j['id'],)).fetchone()[0] for t in ('job_materials','time_entries','staff_updates','calendar_bookings','field_logs','field_sessions','job_photos','recurring_contracts','feedback_requests')} permitted.append({'job_id':j['id'],'service':j['service'],'counts':counts}) timeline=[] for j in jobs:timeline.append({'kind':'job','id':j['id'],'date':j['created_at'],'title':j['service'],'state':j['status']}) for kind,documents in (('quote',quotes),('invoice',invoices)): for d in documents:timeline.append({'kind':kind,'id':d['id'],'date':d['created_at'],'title':d['reference'],'state':d['status'],'amount_pence':d['total_pence']}) for p in payments:timeline.append({'kind':'receipt','id':p['id'],'date':p['paid_on'],'title':p['reference'],'state':'reversed' if p['reversed'] else 'recorded','amount_pence':p['amount_pence']}) for r in records:timeline.append({'kind':'customer','id':str(r['id']),'date':r['created_at'],'title':r['action'],'state':r['reason']}) timeline.sort(key=lambda r:(r['date'],r['kind'],r['id']),reverse=True) missing=[{'job_id':j['id'],'job_version':j['version'],'customer_id':cid,'customer_version':c['version'],'service':j['service'], 'invoices':[{'id':i['id'],'reference':i['reference']} for i in invoices if i['job_id']==j['id'] and i['status']=='issued']} for j in jobs if j['status']=='completed' and not j['workflow']['completed_date'] and not j['workflow']['archived']] from .customer_moves import history as ownership owned_ids=ids|{r['job_id'] for r in links} corrections=[{**r,'plan':json.loads(r['plan']),'result':json.loads(r['result'])} for r in many(db,'SELECT * FROM customer_merge_corrections WHERE alias_id=? OR retained_id=? ORDER BY rowid',(cid,cid))] return {'merge_corrections':corrections,'ownership_history':[ownership(db,jid) for jid in sorted(owned_ids)],'customer':c,'jobs':jobs,'quotes':quotes,'invoices':invoices,'payments':payments,'links':links,'contact_changes':changes, 'record_history':records,'transfers':transfers,'historical_documents':historical,'permitted_records':permitted, 'followups':followups,'timeline':timeline[:1000],'timeline_total':len(timeline),'missing_completion_dates':missing, 'aliases':many(db,"SELECT c.id,c.name FROM customers c JOIN customer_records r ON r.customer_id=c.id WHERE r.canonical_id=? ORDER BY c.id",(cid,)), 'totals':{'jobs':len(jobs),'invoiced_pence':sum(i['total_pence'] for i in invoices if i['status']=='issued'), 'received_pence':sum(p['amount_pence'] for p in payments if not p['reversed']), 'outstanding_pence':sum(i['balance_pence'] for i in invoices)}, 'definitions':'Current associations use exact IDs. Historical document identity is immutable and is shown separately. Issued value excludes void/draft invoices; recorded receipts exclude reversed errors and are not bank-cleared evidence. No duplicate row or retired alias adds financial value.'} EXPORT_FIELDS=('customer_id','state','canonical_customer_id','name','email','phone','address','private_notes', 'created_at_utc','current_jobs','issued_invoice_value_pence','recorded_receipts_pence','outstanding_pence') def csv_text(value): """Preserve Unicode text but prevent common spreadsheet formula execution.""" value=str(value) probe=unicodedata.normalize('NFKC',value).lstrip('\ufeff\u200b\u200c\u200d\u2060 \t\r\n\v\f') if (probe and probe[0] in '=+-@') or value.startswith(('\t','\r','\n')): return "'"+value return value def export_csv(db,q,state): rows=review.directory(db,q,state) if len(rows)>500: review.fail('Narrow this export to at most 500 customer records. No partial file was produced.',422) stream=io.StringIO(newline='');writer=csv.writer(stream,quoting=csv.QUOTE_ALL,lineterminator='\r\n');writer.writerow(EXPORT_FIELDS) invoices=invoice_rows(db) for c in rows: inv=[i for i in invoices if i['customer_id']==c['id']] fields=[c['id'],c['state'],c['canonical_id'] or '',c['name'],c['email'],c['phone'],c['address'],c['notes'],c['created_at']] writer.writerow([*(csv_text(v) for v in fields),c['job_count'],sum(i['total_pence'] for i in inv if i['status']=='issued'),sum(i['paid_pence'] for i in inv),sum(i['balance_pence'] for i in inv)]) return stream.getvalue().encode('utf-8-sig') def mount_customer_reviews(app,store,session,once): @app.get('/api/customers/export') def export(request:Request,q:str=Query('',max_length=180),state:str=Query('active',pattern='^(active|archived|merged|all)$')): s=session(request) with store.tenant(s['business_id']) as db: db.execute('BEGIN');body=export_csv(db,q,state) return Response(body,media_type='text/csv; charset=utf-8',headers={'Content-Disposition':'attachment; filename="customer-records.csv"'}) @app.put('/api/customers/{cid}/private-notes') def notes(cid:str,payload:m.RecordNotes,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db: def act(): c=review.active(db,cid) if c['record_version']!=payload.version: review.fail('Private notes changed. Keep your text and reload before saving.') db.execute('UPDATE customer_records SET notes=? WHERE customer_id=?',(payload.notes,cid)) review.record(db,cid,'notes',s['subject'],'Owner-private notes saved.',advance=True) return review.contact(db,cid) return once(db,request,payload.model_dump(),act) @app.post('/api/customers/{cid}/record-state') def state(cid:str,payload:m.RecordState,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db: def act(): c=review.contact(db,cid) if c['state']=='merged' or c['version']!=payload.version or c['state']==payload.state: review.fail('This state changed or belongs to a retired alias. Reload before reviewing.') if payload.state=='archived': for j in many(db,'SELECT id,status FROM jobs WHERE customer_id=?',(cid,)): if j['status'] not in ('completed','cancelled') or review.work_blockers(db,j['id']): review.fail('Finish or cancel open work and review timers, repeat work and bookings before archiving this customer.') review.pause_sharing(db,cid,cid,s) db.execute('UPDATE customer_records SET state=? WHERE customer_id=?',(payload.state,cid)) db.execute('UPDATE customers SET version=version+1 WHERE id=?',(cid,)) review.record(db,cid,'archive' if payload.state=='archived' else 'restore',s['subject'],payload.reason,advance=True) return review.contact(db,cid) return once(db,request,payload.model_dump(),act) @app.get('/api/customer-reviews/duplicates') def duplicates(request:Request): s=session(request) with store.tenant(s['business_id']) as db: db.execute('BEGIN');return review.suggestions(db) @app.post('/api/customer-reviews/pairs') def pair(payload:m.PairDecision,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db:return once(db,request,payload.model_dump(),lambda:review.decide_pair(db,payload,s['subject'])) @app.get('/api/customer-reviews/pairs/{pair_id}/history') def pair_history(pair_id:str,request:Request): s=session(request) with store.tenant(s['business_id']) as db: if not one(db,'SELECT id FROM customer_pair_decisions WHERE id=?',(pair_id,)):review.fail('Pair decision not found in this business.',404) return [{**r,'snapshot':json.loads(r['snapshot'])} for r in many(db,'SELECT * FROM customer_pair_history WHERE pair_id=? ORDER BY version',(pair_id,))] @app.post('/api/customer-reviews/transfer-preview') def preview(payload:m.TransferSelection,request:Request): with store.registry() as registry: registry.execute('BEGIN IMMEDIATE') s=session(request,csrf=True) if s['role']!='owner':review.fail('Only the current owner may review customer corrections.',403) with store.tenant(s['business_id']) as db: db.execute('BEGIN IMMEDIATE');return review.transfer_plan(db,payload) @app.post('/api/customer-reviews/transfers') def transfer(payload:m.TransferConfirm,request:Request): from uuid import UUID with store.registry() as registry: registry.execute('BEGIN IMMEDIATE') s=session(request,csrf=True) if s['role']!='owner':review.fail('Only the current owner may confirm customer corrections.',403) with store.tenant(s['business_id']) as db: return once(db,request,payload.model_dump(),lambda:review.execute_transfer(db,payload,s,UUID(request.headers['Idempotency-Key']).hex)) @app.get('/api/customer-reviews/transfers/{tid}') def saved_transfer(tid:str,request:Request): s=session(request) with store.tenant(s['business_id']) as db: row=one(db,'SELECT * FROM customer_transfers WHERE id=?',(tid,)) if not row:review.fail('Transfer not found in this business.',404) return {**row,'plan':json.loads(row['plan']),'result':json.loads(row['result'])} @app.post('/api/customer-reviews/completion-preview') def completion_preview(payload:m.CompletionPreview,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db: db.execute('BEGIN');return review.completion_plan(db,payload) @app.post('/api/customer-reviews/completions') def completion(payload:m.CompletionConfirm,request:Request): s=session(request,csrf=True) with store.tenant(s['business_id']) as db:return once(db,request,payload.model_dump(),lambda:review.complete_dates(db,payload,s)) @app.get('/api/customer-reviews/merges/{tid}/correction') def merge_correction_context(tid:str,request:Request): from .merge_corrections import context with store.registry() as registry: registry.execute('BEGIN IMMEDIATE');s=session(request) if s['role']!='owner':review.fail('Only the current owner may review merge corrections.',403) with store.tenant(s['business_id']) as db: db.execute('BEGIN');return context(db,tid) @app.post('/api/customer-reviews/merge-correction-preview') def merge_correction_preview(payload:m.MergeCorrectionSelection,request:Request): from .merge_corrections import plan with store.registry() as registry: registry.execute('BEGIN IMMEDIATE');s=session(request,csrf=True) if s['role']!='owner':review.fail('Only the current owner may review merge corrections.',403) with store.tenant(s['business_id']) as db: db.execute('BEGIN IMMEDIATE');return plan(db,payload,s) @app.post('/api/customer-reviews/merge-corrections') def merge_correction_confirm(payload:m.MergeCorrectionConfirm,request:Request): from uuid import UUID from .merge_corrections import execute,check_receipt_access with store.registry() as registry: registry.execute('BEGIN IMMEDIATE');s=session(request,csrf=True) if s['role']!='owner':review.fail('Only the current owner may confirm merge corrections.',403) with store.tenant(s['business_id']) as db: def operation_id(): try:return UUID(request.headers.get('Idempotency-Key','')).hex except ValueError:review.fail('A unique operation ID is required.',400) tid=operation_id() return once(db,request,payload.model_dump(),lambda:execute(db,payload,s,tid),check=lambda:check_receipt_access(db,tid,s))