import csv import hashlib import hmac import io import json import logging import os import re import secrets import sqlite3 import threading import time from collections import defaultdict, deque from datetime import datetime, timedelta from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse from db import ROOT, audit, connect, init, now, one, password_hash, password_ok, rows from engine import create_order, phone, receive, reply, validated_items from scheduling import IST, next_delivery # .env values are only read by the backend; never served as static files. env_path=ROOT/'.env' if env_path.exists(): for line in env_path.read_text(encoding='utf-8').splitlines(): if line.strip() and not line.lstrip().startswith('#') and '=' in line: k,v=line.split('=',1) os.environ.setdefault(k.strip(),v.strip().strip('"').strip("'")) SESSIONS={} RATE=defaultdict(deque) LOCK=threading.Lock() FIELDS={ 'routes':['name','weekday','cutoff_days','cutoff_time','departure_time','active'], 'areas':['name','city','aliases'], 'customers':['name','shop','phone','alternate_phone','area_id','address','landmark','status','notes'], 'products':['name','sku','category','unit','price','aliases','active'], 'vehicles':['name','registration','active'], 'drivers':['name','phone'], 'whatsapp_numbers':['label','phone_number_id','token_env','active'], 'holidays':['date','reason'], 'message_templates':['name','body'], 'keyword_flows':['name','keywords','match_mode','action','response','priority','active','number_id'], } REQUIRED={'routes':['name','weekday'],'areas':['name'],'customers':['name','phone'],'products':['name','sku'],'vehicles':['name','registration'],'drivers':['name','phone'],'whatsapp_numbers':['label','phone_number_id'],'holidays':['date'],'message_templates':['name','body']} REQUIRED['keyword_flows']=['name','keywords','action'] def order_rows(db): result=rows(db,"SELECT o.*,c.name AS customer,c.shop,c.phone,c.address,c.landmark,a.name AS area,r.name AS route,r.weekday,COALESCE(s.position,9999) AS position,v.name AS vehicle,d.name AS driver,ds.vehicle_id,ds.driver_id,dl.status AS delivery_status FROM orders o JOIN customers c ON c.id=o.customer_id JOIN areas a ON a.id=o.area_id JOIN routes r ON r.id=o.route_id LEFT JOIN route_stops s ON s.area_id=o.area_id AND s.route_id=o.route_id LEFT JOIN delivery_schedules ds ON ds.route_id=o.route_id AND ds.date=o.delivery_date LEFT JOIN vehicles v ON v.id=ds.vehicle_id LEFT JOIN drivers d ON d.id=ds.driver_id LEFT JOIN deliveries dl ON dl.order_id=o.id WHERE o.business_id=1 ORDER BY o.delivery_date,s.position,o.id DESC") for order in result: order['items']=rows(db,'SELECT i.*,p.name AS product FROM order_items i JOIN products p ON p.id=i.product_id WHERE i.order_id=?',(order['id'],)) order['total']=round(sum(i['quantity']*i['price'] for i in order['items']),2) return result def validate(table,data): unknown=set(data)-set(FIELDS[table]) if unknown: raise ValueError('Unsupported fields: '+', '.join(unknown)) for field in REQUIRED[table]: if field not in data or data[field] is None or str(data[field]).strip()=='': raise ValueError(field+' is required') for k,v in data.items(): if isinstance(v,str) and len(v)>10000: raise ValueError(k+' is too long') if 'phone' in data: data['phone']=phone(data['phone']) if data.get('alternate_phone'): data['alternate_phone']=phone(data['alternate_phone']) if table=='routes': if not 0<=int(data['weekday'])<=6: raise ValueError('Invalid weekday') if not 0<=int(data.get('cutoff_days',1))<=6: raise ValueError('Cutoff offset must be 0–6 days') for k in ('cutoff_time','departure_time'): if k in data and not re.fullmatch(r'(?:[01]\d|2[0-3]):[0-5]\d',str(data[k])): raise ValueError('Use HH:MM times') if table=='products': if float(data.get('price',0))<0: raise ValueError('Price cannot be negative') if 'date' in data: datetime.strptime(data['date'],'%Y-%m-%d') if table=='whatsapp_numbers': if data['phone_number_id']!='local' and not str(data['phone_number_id']).isdigit(): raise ValueError('Phone number ID must be numeric') if not re.fullmatch(r'[A-Z][A-Z0-9_]*',data.get('token_env','WHATSAPP_ACCESS_TOKEN')): raise ValueError('Use an environment variable name for token_env') if table=='message_templates': if data['name']!='confirmation': raise ValueError('V1 supports the confirmation template') try: data['body'].format(customer_name='Test',order_id=1,area='Area',delivery_day='Monday',delivery_date='2026-10-12') except (KeyError,ValueError,IndexError) as e: raise ValueError('Unsupported template placeholder') from e if table=='keyword_flows': if data['action'] not in ('reply','order','review'):raise ValueError('Invalid flow action') if data.get('match_mode','any') not in ('any','all','exact'):raise ValueError('Invalid matching mode') if data['action']=='reply' and not str(data.get('response','')).strip():raise ValueError('Reply flow needs a response') if len(data.get('response',''))>4096:raise ValueError('Reply is too long') class Handler(BaseHTTPRequestHandler): server_version='RouteFlow/1.0' def log_message(self,fmt,*args): logging.info('%s %s',self.command,self.path.split('?')[0]) def respond(self,data,status=200,content_type='application/json',cookie=None): payload=json.dumps(data,ensure_ascii=False).encode() if content_type=='application/json' else (data.encode() if isinstance(data,str) else data) self.send_response(status) self.send_header('Content-Type',content_type+'; charset=utf-8') self.send_header('Content-Length',str(len(payload))) self.send_header('Cache-Control','no-store') self.send_header('X-Content-Type-Options','nosniff') self.send_header('X-Frame-Options','DENY') self.send_header('Referrer-Policy','same-origin') self.send_header('Content-Security-Policy',"default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; base-uri 'none'; frame-ancestors 'none'") if cookie: self.send_header('Set-Cookie',cookie) self.end_headers() self.wfile.write(payload) def body(self): length=int(self.headers.get('Content-Length','0')) if length<0 or length>1024*1024: raise ValueError('Request too large') return self.rfile.read(length) def limited(self,limit=120): key=(self.client_address[0],self.path.split('?')[0]) with LOCK: q=RATE[key] stamp=time.monotonic() while q and q[0]=limit: return True q.append(stamp) return False def session(self): cookie=SimpleCookie() cookie.load(self.headers.get('Cookie','')) token=cookie.get('routeflow_session') with LOCK: session=SESSIONS.get(token.value if token else '') if session and session['expires']>time.time(): return session return None def do_GET(self): self.handle_request('GET') def do_POST(self): self.handle_request('POST') def do_PUT(self): self.handle_request('PUT') def do_DELETE(self): self.handle_request('DELETE') def handle_request(self,method): try: path=urlparse(self.path).path query=parse_qs(urlparse(self.path).query) if path in ('/','/app.js','/style.css','/linked-ui.js','/linked.css') and method=='GET': name={'/':'index.html','/app.js':'app.js','/style.css':'style.css','/linked-ui.js':'linked-ui.js','/linked.css':'linked.css'}[path] types={'index.html':'text/html','app.js':'text/javascript','style.css':'text/css','linked-ui.js':'text/javascript','linked.css':'text/css'} return self.respond((ROOT/'static'/name).read_bytes(),content_type=types[name]) if path=='/health' and method=='GET': return self.respond({'status':'ok'}) if path=='/internal/linked-event' and method=='POST': from linked import bridge_secret,ingest if self.client_address[0]!='127.0.0.1' or not hmac.compare_digest(self.headers.get('X-Bridge-Secret',''),bridge_secret()):return self.respond({'error':'Unauthorized bridge'},403) payload=json.loads(self.body()) with connect() as db:ingest(db,payload) return self.respond({'accepted':True}) if path=='/webhook/whatsapp': if method=='GET': verify=os.environ.get('WHATSAPP_VERIFY_TOKEN','') if verify and query.get('hub.mode')==['subscribe'] and hmac.compare_digest(query.get('hub.verify_token',[''])[0],verify): return self.respond(query.get('hub.challenge',[''])[0],content_type='text/plain') return self.respond({'error':'Verification failed'},403) if method!='POST': return self.respond({'error':'Method not allowed'},405) raw=self.body() secret=os.environ.get('WHATSAPP_APP_SECRET','') expected='sha256='+hmac.new(secret.encode(),raw,hashlib.sha256).hexdigest() if not secret or not hmac.compare_digest(expected,self.headers.get('X-Hub-Signature-256','')): return self.respond({'error':'Invalid webhook signature'},403) json.loads(raw) with connect() as db: db.execute('INSERT OR IGNORE INTO webhook_events(body_hash,raw_body,created_at) VALUES(?,?,?)',(hashlib.sha256(raw).hexdigest(),raw.decode(),now())) return self.respond({'accepted':True}) if self.limited(10 if path=='/api/login' else 240): return self.respond({'error':'Rate limit exceeded'},429) if path=='/api/login' and method=='POST': data=json.loads(self.body()) with connect() as db: user=one(db,'SELECT * FROM users WHERE username=?',(data.get('username',''),)) if not user or not password_ok(data.get('password',''),user['password_hash']): return self.respond({'error':'Incorrect username/password'},401) token=secrets.token_urlsafe(32) session={'user':{k:user[k] for k in ('id','username','role','driver_id')},'csrf':secrets.token_urlsafe(24),'expires':time.time()+8*3600} with LOCK: for old in list(SESSIONS): if SESSIONS[old]['expires']4096:raise ValueError('Reply must contain 1–4096 characters') from linked import ensure_number number_id=ensure_number(db) db.execute('INSERT OR IGNORE INTO conversations(business_id,customer_id,number_id) VALUES(1,?,?)',(chat['customer_id'],number_id)) conv=one(db,'SELECT id FROM conversations WHERE customer_id=? AND number_id=?',(chat['customer_id'],number_id)) mid=reply(db,conv['id'],body) db.execute('UPDATE outbox SET payload_json=? WHERE message_id=?',(json.dumps({'text':body,'manual':True}),mid)) audit(db,'Manual Linked Reply Queued','messages',mid,actor=actor) return {'ok':True} if path=='/api/flow-test' and method=='POST': from linked import matched_flow return {'match':matched_flow(db,str(data.get('text','')),data.get('number_id'))} if path=='/api/driver' and method=='GET': orders=order_rows(db) driver_id=session['user']['driver_id'] return {'orders':[o for o in orders if o['delivery_date']==datetime.now(IST).date().isoformat() and o['status']!='Cancelled' and (session['user']['role']!='driver' or o['driver_id']==driver_id)]} table=path.removeprefix('/api/') if table in FIELDS: if method=='POST': validate(table,data) if table=='customers' and data.get('area_id'): if not one(db,'SELECT id FROM areas WHERE id=? AND business_id=1',(data['area_id'],)): raise ValueError('Invalid area') cols=['business_id']+list(data) eid=db.execute('INSERT INTO '+table+'('+','.join(cols)+') VALUES('+','.join('?' for _ in cols)+')',[1]+list(data.values())).lastrowid elif method=='PUT': if not data: raise ValueError('No fields supplied') eid=int(query.get('id',['0'])[0]) existing=one(db,'SELECT * FROM '+table+' WHERE id=? AND business_id=1',(eid,)) if not existing: raise ValueError('Record not found') merged={k:existing[k] for k in FIELDS[table]} merged.update(data); validate(table,merged) db.execute('UPDATE '+table+' SET '+','.join(k+'=?' for k in data)+' WHERE id=? AND business_id=1',list(data.values())+[eid]) elif method=='DELETE': eid=int(query.get('id',['0'])[0]) db.execute('DELETE FROM '+table+' WHERE id=? AND business_id=1',(eid,)) else: raise ValueError('Use /api/state to read records') audit(db,'Admin '+method,table,eid,data,actor) return {'ok':True,'id':eid} if path=='/api/stops' and method=='POST': route_id=int(data['route_id']) ids=[int(a) for a in data['area_ids']] if len(ids)!=len(set(ids)): raise ValueError('Duplicate area in stops') if not one(db,'SELECT id FROM routes WHERE id=? AND business_id=1',(route_id,)): raise ValueError('Route not found') for aid in ids: if not one(db,'SELECT id FROM areas WHERE id=? AND business_id=1',(aid,)): raise ValueError('Area not found') used=one(db,'SELECT route_id FROM route_stops WHERE area_id=?',(aid,)) if used and used['route_id']!=route_id: raise ValueError('Area already assigned to another route; remove it there first') db.execute('DELETE FROM route_stops WHERE route_id=?',(route_id,)) for idx,aid in enumerate(ids): db.execute('INSERT INTO route_stops(route_id,area_id,position) VALUES(?,?,?)',(route_id,aid,idx+1)) audit(db,'Route Stops Updated','routes',route_id,data,actor) return {'ok':True} if path=='/api/orders' and method=='POST': number_id=None; message_id=None if data.get('review_message_id'): reviewed=one(db,"SELECT m.id,c.number_id,c.customer_id,c.id AS conversation_id FROM messages m JOIN conversations c ON c.id=m.conversation_id WHERE m.id=? AND m.direction='in' AND m.state IN ('review','human')",(int(data['review_message_id']),)) if not reviewed or reviewed['customer_id']!=int(data['customer_id']): raise ValueError('Review message must belong to this customer and be unresolved') number_id=reviewed['number_id'];message_id=reviewed['id'] oid=create_order(db,int(data['customer_id']),data['items'],number_id=number_id,message_id=message_id,notes=data.get('notes',''),actor=actor,override=data.get('override'),allow_duplicate=bool(data.get('allow_duplicate',False))) if data.get('review_message_id'): mid=int(data['review_message_id']) db.execute("UPDATE messages SET state='review_resolved' WHERE id=?",(mid,)) db.execute('UPDATE notifications SET resolved=1 WHERE message_id=?',(mid,)) db.execute('UPDATE conversations SET pending_json=NULL WHERE id=?',(reviewed['conversation_id'],)) return {'id':oid} if path=='/api/orders' and method=='PUT': oid=int(data['id']); existing=one(db,'SELECT * FROM orders WHERE id=? AND business_id=1',(oid,)) if not existing: raise ValueError('Order not found') allowed={'id','customer_id','area_id','route_id','delivery_date','status','payment_status','notes','items'} if set(data)-allowed: raise ValueError('Unsupported order fields') for field,target in (('customer_id','customers'),('area_id','areas'),('route_id','routes')): if field in data and not one(db,'SELECT id FROM '+target+' WHERE id=? AND business_id=1',(data[field],)): raise ValueError('Invalid '+field) if 'status' in data and data['status'] not in ('New','Needs Confirmation','Confirmed','Scheduled','Out for Delivery','Delivered','Cancelled','Rescheduled'): raise ValueError('Invalid order status') if 'payment_status' in data and data['payment_status'] not in ('Pending','Paid','Credit'): raise ValueError('Invalid payment status') if 'delivery_date' in data: datetime.strptime(data['delivery_date'],'%Y-%m-%d') if 'route_id' in data and 'delivery_date' not in data: data['delivery_date']=next_delivery(db,int(data['route_id'])) if 'items' in data: items=validated_items(db,data['items']) db.execute('DELETE FROM order_items WHERE order_id=?',(oid,)) for i in items: db.execute('INSERT INTO order_items(order_id,product_id,quantity,unit,price) VALUES(?,?,?,?,?)',(oid,i['product_id'],i['quantity'],i['unit'],i['price'])) fp=hashlib.sha256(json.dumps(sorted([(i['product_id'],i['quantity'],i['unit']) for i in items]),sort_keys=True).encode()).hexdigest() db.execute('UPDATE orders SET fingerprint=? WHERE id=?',(fp,oid)) values={k:v for k,v in data.items() if k not in ('id','items')} if values: db.execute('UPDATE orders SET '+','.join(k+'=?' for k in values)+' WHERE id=?',list(values.values())+[oid]) if data.get('status') in ('Delivered','Cancelled'): db.execute('UPDATE deliveries SET status=?,updated_at=? WHERE order_id=?',(data['status'],now(),oid)) audit(db,'Order Edited','orders',oid,data,actor) return {'ok':True} if path=='/api/simulate' and method=='POST': if data.get('provider'): from adapters import wa_akg,wechaty adapter={'wa-akg':wa_akg,'wechaty':wechaty}.get(data['provider']) if not adapter: raise ValueError('Unknown provider') data=adapter(data['payload']) number=int(data.get('number_id',1)) if not one(db,"SELECT id FROM whatsapp_numbers WHERE id=? AND phone_number_id='local'",(number,)): raise ValueError('Simulator can only use the local number') mid=receive(db,number,data['phone'],data['body'],data.get('external_id'),data.get('kind','text')) return {'message_id':mid,'queued':True} if path=='/api/takeover' and method=='POST': cid=int(data['conversation_id']) db.execute('UPDATE conversations SET takeover=?,pending_json=NULL WHERE id=?',(int(bool(data['enabled'])),cid)) audit(db,'Human Takeover Changed','conversations',cid,data,actor) return {'ok':True} if path=='/api/reply' and method=='POST': cid=int(data['conversation_id']) if not one(db,'SELECT id FROM conversations WHERE id=? AND business_id=1',(cid,)): raise ValueError('Conversation not found') body=str(data['body']).strip() if not body or len(body)>4096: raise ValueError('Reply must contain 1–4096 characters') mid=reply(db,cid,body) payload={'text':body,'manual':True} if data.get('template'): payload['template']=data['template'] db.execute('UPDATE outbox SET payload_json=? WHERE message_id=?',(json.dumps(payload),mid)) audit(db,'Manual Reply Queued','messages',mid,actor=actor) return {'ok':True} if path=='/api/assign' and method=='POST': rid=int(data['route_id']); date=data['date']; datetime.strptime(date,'%Y-%m-%d') if not one(db,'SELECT id FROM routes WHERE id=? AND business_id=1',(rid,)): raise ValueError('Route not found') for key,table in (('vehicle_id','vehicles'),('driver_id','drivers')): if data.get(key) and not one(db,'SELECT id FROM '+table+' WHERE id=? AND business_id=1',(data[key],)): raise ValueError('Invalid '+key) if data.get('vehicle_id') and not one(db,'SELECT id FROM vehicles WHERE id=? AND active=1',(data['vehicle_id'],)): raise ValueError('Vehicle is unavailable') db.execute('INSERT INTO delivery_schedules(route_id,date,vehicle_id,driver_id,departed) VALUES(?,?,?,?,?) ON CONFLICT(route_id,date) DO UPDATE SET vehicle_id=excluded.vehicle_id,driver_id=excluded.driver_id,departed=excluded.departed',(rid,date,data.get('vehicle_id'),data.get('driver_id'),int(bool(data.get('departed'))))) audit(db,'Route Assignment','routes',rid,data,actor) return {'ok':True} if path=='/api/delivery' and method=='POST': oid=int(data['order_id']) order=next((o for o in order_rows(db) if o['id']==oid),None) if not order: raise ValueError('Order not found') if session['user']['role']=='driver' and order['driver_id']!=session['user']['driver_id']: raise ValueError('Order is not assigned to you') status=data['status'] if status not in ('Delivered','Customer Not Available','Cancelled','Payment Pending','Reschedule'): raise ValueError('Invalid delivery status') db.execute('UPDATE deliveries SET status=?,notes=?,updated_at=? WHERE order_id=?',(status,data.get('notes',''),now(),oid)) if status in ('Delivered','Cancelled'): db.execute('UPDATE orders SET status=? WHERE id=?',(status,oid)) if status=='Payment Pending': db.execute("UPDATE orders SET payment_status='Pending' WHERE id=?",(oid,)) if status=='Reschedule': date=data.get('date') if not date: raise ValueError('Choose a new delivery date') datetime.strptime(date,'%Y-%m-%d') db.execute("UPDATE orders SET status='Rescheduled',delivery_date=? WHERE id=?",(date,oid)) audit(db,'Delivery Updated','orders',oid,data,actor) return {'ok':True} if path=='/api/settings' and method=='POST': if not set(data)<= {'duplicate_minutes','automation','summary_time'}: raise ValueError('Unsupported setting') if 'duplicate_minutes' in data and not 1<=int(data['duplicate_minutes'])<=1440: raise ValueError('Duplicate window must be 1–1440 minutes') if 'automation' in data and data['automation'] not in ('true','false'): raise ValueError('Invalid automation setting') if 'summary_time' in data and not re.fullmatch(r'(?:[01]\d|2[0-3]):[0-5]\d',data['summary_time']): raise ValueError('Use HH:MM summary time') for k,v in data.items(): db.execute('INSERT INTO settings VALUES(1,?,?) ON CONFLICT(business_id,key) DO UPDATE SET value=excluded.value',(k,str(v))) audit(db,'Settings Updated',detail=data,actor=actor) return {'ok':True} if path=='/api/users' and method=='POST': if session['user']['role']!='admin': raise ValueError('Admin required') if data['role'] not in ('admin','staff','driver'): raise ValueError('Invalid role') if len(data['password'])<12: raise ValueError('Password needs at least 12 characters') if data['role']=='driver' and not one(db,'SELECT id FROM drivers WHERE id=?',(data.get('driver_id'),)): raise ValueError('Assign a driver profile') uid=db.execute('INSERT INTO users(business_id,username,password_hash,role,driver_id) VALUES(1,?,?,?,?)',(data['username'],password_hash(data['password']),data['role'],data.get('driver_id'))).lastrowid audit(db,'User Created','users',uid,{'role':data['role']},actor) return {'id':uid} if path=='/api/resolve' and method=='POST': db.execute('UPDATE notifications SET resolved=1 WHERE id=?',(int(data['id']),)) audit(db,'Notification Resolved','notifications',int(data['id']),actor=actor) return {'ok':True} if path=='/api/retry' and method=='POST': if data.get('webhook_id'): db.execute("UPDATE webhook_events SET state='pending',error=NULL WHERE id=? AND state='failed'",(int(data['webhook_id']),)) elif data.get('outbox_id'): db.execute("UPDATE outbox SET state='pending',next_attempt_at=? WHERE id=? AND state='failed'",(now(),int(data['outbox_id']))) audit(db,'Queue Retry Requested',detail=data,actor=actor) return {'ok':True} if path=='/api/customer-history' and method=='GET': cid=int(query['id'][0]) return {'customer':one(db,'SELECT * FROM customers WHERE id=? AND business_id=1',(cid,)),'orders':[o for o in order_rows(db) if o['customer_id']==cid],'frequent_products':rows(db,'SELECT p.name,SUM(i.quantity) AS quantity,COUNT(*) AS orders FROM order_items i JOIN products p ON p.id=i.product_id JOIN orders o ON o.id=i.order_id WHERE o.customer_id=? AND o.status!=\'Cancelled\' GROUP BY p.id ORDER BY orders DESC',(cid,))} if path=='/api/demo' and method=='POST': if one(db,'SELECT id FROM routes LIMIT 1'): raise ValueError('Demo data only loads into an empty business') from seed import seed seed(db) audit(db,'Demo Data Loaded',actor=actor) return {'ok':True} raise ValueError('Unknown endpoint') def main(): logging.basicConfig(level=logging.INFO,format='%(asctime)s %(levelname)s %(message)s') init() from worker import start start() host=os.environ.get('HOST','127.0.0.1'); port=int(os.environ.get('PORT','8765')) print(f'RouteFlow dashboard: http://{host}:{port}',flush=True) ThreadingHTTPServer((host,port),Handler).serve_forever() if __name__=='__main__': main()