add operator review workflow
This commit is contained in:
Marco0300
+154
-3
@@ -55,7 +55,7 @@ def connect(db_path: str) -> sqlite3.Connection:
|
||||
db = sqlite3.connect(db_path); db.row_factory = sqlite3.Row; db.execute("PRAGMA foreign_keys = ON"); db.executescript(SCHEMA.read_text())
|
||||
# Upgrade databases created by Phase 1/2 without destroying data.
|
||||
cols = {r[1] for r in db.execute("PRAGMA table_info(businesses)")}
|
||||
for col, definition in (("verified", "INTEGER NOT NULL DEFAULT 0"), ("verified_at", "TEXT"), ("updated_at", "TEXT"), ("province", "TEXT NOT NULL DEFAULT ''"), ("city", "TEXT NOT NULL DEFAULT ''"), ("suburb", "TEXT NOT NULL DEFAULT ''"), ("merge_status", "TEXT NOT NULL DEFAULT 'active'"), ("merged_into_id", "INTEGER")):
|
||||
for col, definition in (("verified", "INTEGER NOT NULL DEFAULT 0"), ("verified_at", "TEXT"), ("updated_at", "TEXT"), ("province", "TEXT NOT NULL DEFAULT ''"), ("city", "TEXT NOT NULL DEFAULT ''"), ("suburb", "TEXT NOT NULL DEFAULT ''"), ("merge_status", "TEXT NOT NULL DEFAULT 'active'"), ("merged_into_id", "INTEGER"), ("review_status", "TEXT NOT NULL DEFAULT 'pending'"), ("assigned_to", "TEXT NOT NULL DEFAULT ''"), ("review_metadata_json", "TEXT NOT NULL DEFAULT '{}'")):
|
||||
if col not in cols: db.execute(f"ALTER TABLE businesses ADD COLUMN {col} {definition}")
|
||||
db.execute("UPDATE businesses SET updated_at=COALESCE(updated_at,created_at) WHERE updated_at IS NULL")
|
||||
db.execute("INSERT OR IGNORE INTO organizations (id,name) VALUES (?,?)", (ORGANIZATION_ID, "Demo organization"))
|
||||
@@ -111,6 +111,117 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
for key, table in tables.items():
|
||||
result[key]=[row_json(r) for r in db.execute(f"SELECT * FROM {table} WHERE business_id=? AND organization_id=? ORDER BY id",(bid,org))]
|
||||
return result
|
||||
def _bounded_filters(self, value):
|
||||
def walk(item, depth=0):
|
||||
if depth > 4: raise ValueError("filters_too_deep")
|
||||
if isinstance(item, dict):
|
||||
if len(item) > 30: raise ValueError("filters_too_large")
|
||||
return {str(k)[:80]: walk(v, depth + 1) for k, v in item.items()}
|
||||
if isinstance(item, list):
|
||||
if len(item) > 50: raise ValueError("filters_too_large")
|
||||
return [walk(v, depth + 1) for v in item]
|
||||
if isinstance(item, str):
|
||||
if len(item) > 500: raise ValueError("filter_value_too_large")
|
||||
return item
|
||||
if item is None or isinstance(item, (bool, int, float)):
|
||||
return item
|
||||
raise ValueError("invalid_filters")
|
||||
if not isinstance(value, dict): raise ValueError("invalid_filters")
|
||||
result = walk(value)
|
||||
if len(json.dumps(result, separators=(",", ":"), ensure_ascii=False).encode()) > 8192: raise ValueError("filters_too_large")
|
||||
return result
|
||||
|
||||
def _saved_filter_json(self, row):
|
||||
item = row_json(row)
|
||||
try: item["filters"] = json.loads(item.pop("filters_json") or "{}")
|
||||
except (TypeError, ValueError): item["filters"] = {}
|
||||
return item
|
||||
|
||||
def list_saved_filters(self, db, user):
|
||||
rows = db.execute("SELECT * FROM saved_filters WHERE organization_id=? AND user_id=? ORDER BY updated_at DESC,id DESC", (user["organization_id"], user["id"])).fetchall()
|
||||
return self.send_json(200, {"organization_id": user["organization_id"], "items": [self._saved_filter_json(r) for r in rows]})
|
||||
|
||||
def save_filter(self, payload, db, user):
|
||||
name = str(payload.get("name", "")).strip()
|
||||
raw = payload.get("filters", payload.get("filter", payload.get("filters_json", {})))
|
||||
if not name or len(name) > 120: return self.send_json(400, {"error": "invalid_saved_filter"})
|
||||
try: filters = self._bounded_filters(raw)
|
||||
except ValueError as exc: return self.send_json(400, {"error": str(exc)})
|
||||
try:
|
||||
cur = db.execute("INSERT INTO saved_filters(organization_id,user_id,name,filters_json) VALUES(?,?,?,?)", (user["organization_id"], user["id"], name, json.dumps(filters, sort_keys=True, separators=(",", ":"))))
|
||||
except sqlite3.IntegrityError: return self.send_json(409, {"error": "duplicate_saved_filter"})
|
||||
self.audit(db, user, "saved_filter.created", str(cur.lastrowid)); db.commit()
|
||||
return self.send_json(201, self._saved_filter_json(db.execute("SELECT * FROM saved_filters WHERE id=?", (cur.lastrowid,)).fetchone()))
|
||||
|
||||
def update_saved_filter(self, fid, payload, db, user):
|
||||
row = db.execute("SELECT * FROM saved_filters WHERE id=? AND organization_id=? AND user_id=?", (fid, user["organization_id"], user["id"])).fetchone()
|
||||
if not row: return self.send_json(404, {"error": "not_found"})
|
||||
fields = []; values = []
|
||||
if "name" in payload:
|
||||
name = str(payload["name"]).strip()
|
||||
if not name or len(name) > 120: return self.send_json(400, {"error": "invalid_saved_filter"})
|
||||
fields.append("name=?"); values.append(name)
|
||||
if any(k in payload for k in ("filters", "filter", "filters_json")):
|
||||
try: filters = self._bounded_filters(payload.get("filters", payload.get("filter", payload.get("filters_json"))))
|
||||
except ValueError as exc: return self.send_json(400, {"error": str(exc)})
|
||||
fields.append("filters_json=?"); values.append(json.dumps(filters, sort_keys=True, separators=(",", ":")))
|
||||
if not fields: return self.send_json(400, {"error": "no_changes"})
|
||||
values += [fid, user["organization_id"], user["id"]]
|
||||
try: db.execute("UPDATE saved_filters SET " + ",".join(fields) + ",updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=? AND user_id=?", values)
|
||||
except sqlite3.IntegrityError: return self.send_json(409, {"error": "duplicate_saved_filter"})
|
||||
self.audit(db, user, "saved_filter.updated", str(fid)); db.commit()
|
||||
return self.send_json(200, self._saved_filter_json(db.execute("SELECT * FROM saved_filters WHERE id=?", (fid,)).fetchone()))
|
||||
|
||||
def delete_saved_filter(self, fid, db, user):
|
||||
row = db.execute("SELECT id FROM saved_filters WHERE id=? AND organization_id=? AND user_id=?", (fid, user["organization_id"], user["id"])).fetchone()
|
||||
if not row: return self.send_json(404, {"error": "not_found"})
|
||||
db.execute("DELETE FROM saved_filters WHERE id=? AND organization_id=? AND user_id=?", (fid, user["organization_id"], user["id"]))
|
||||
self.audit(db, user, "saved_filter.deleted", str(fid)); db.commit()
|
||||
return self.send_json(200, {"ok": True, "id": fid})
|
||||
|
||||
def _review_item(self, row, db, org):
|
||||
item = row_json(row)
|
||||
suppressed = is_suppressed(dict(row), [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=?", (org,))])
|
||||
merged = row["merge_status"] == "merged"
|
||||
item.update({"suppressed": bool(suppressed), "merged": bool(merged), "outreach_eligible": not suppressed and not merged and row["review_status"] not in ("rejected",), "review_flags": [x for x, yes in (("suppressed", suppressed), ("merged", merged), ("rejected", row["review_status"] == "rejected")) if yes]})
|
||||
return item
|
||||
|
||||
def review_queue(self, db, org, query):
|
||||
try:
|
||||
limit = int((query.get("page_size") or [50])[0]); offset = int((query.get("offset") or [0])[0])
|
||||
if limit < 1 or limit > 100 or offset < 0: raise ValueError
|
||||
except (ValueError, TypeError): return self.send_json(400, {"error": "invalid_pagination"})
|
||||
where, params = ["b.organization_id=?"], [org]
|
||||
status = (query.get("review_status") or query.get("status") or [""])[0].strip()
|
||||
if status:
|
||||
if status not in {"pending", "verified", "rejected", "assigned"}: return self.send_json(400, {"error": "invalid_review_status"})
|
||||
where.append("b.review_status=?"); params.append(status)
|
||||
for key, op in (("score_min", ">="), ("score_max", "<=")):
|
||||
raw = (query.get(key) or [""])[0]
|
||||
if raw:
|
||||
try: value = int(raw)
|
||||
except ValueError: return self.send_json(400, {"error": "invalid_score"})
|
||||
if value < 0 or value > 100: return self.send_json(400, {"error": "invalid_score"})
|
||||
where.append("b.score" + op + "?"); params.append(value)
|
||||
priority = (query.get("priority") or query.get("priority_band") or [""])[0].strip()
|
||||
if priority:
|
||||
bounds = {"high": (70, 100), "medium": (40, 69), "low": (0, 39)}
|
||||
if priority not in bounds: return self.send_json(400, {"error": "invalid_priority"})
|
||||
where += ["b.score BETWEEN ? AND ?"]; params += list(bounds[priority])
|
||||
website = (query.get("website_state") or query.get("website_class") or [""])[0].strip()
|
||||
if website: where.append("b.website_class=?"); params.append(website)
|
||||
freshness = (query.get("freshness_days") or [""])[0]
|
||||
if freshness:
|
||||
try: days = int(freshness)
|
||||
except ValueError: return self.send_json(400, {"error": "invalid_freshness"})
|
||||
if days < 0 or days > 3650: return self.send_json(400, {"error": "invalid_freshness"})
|
||||
where.append("b.updated_at >= datetime('now', ?)"); params.append(f"-{days} days")
|
||||
source_health = (query.get("source_health") or [""])[0].strip()
|
||||
if source_health:
|
||||
where.append("EXISTS (SELECT 1 FROM sources s WHERE s.organization_id=b.organization_id AND s.health_status=?)"); params.append(source_health)
|
||||
rows = db.execute("SELECT b.* FROM businesses b WHERE " + " AND ".join(where) + " ORDER BY b.score DESC,b.updated_at DESC,b.id DESC LIMIT ? OFFSET ?", params + [limit + 1, offset]).fetchall()
|
||||
return self.send_json(200, {"organization_id": org, "items": [self._review_item(r, db, org) for r in rows[:limit]], "limit": limit, "offset": offset, "has_more": len(rows) > limit})
|
||||
|
||||
def do_GET(self):
|
||||
parsed=urlparse(self.path); path=parsed.path.rstrip("/")
|
||||
if path=="/api/v1/health/live": return self.send_json(200,{"status":"ok","organization_id":ORGANIZATION_ID})
|
||||
@@ -124,7 +235,12 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
if user["role"] not in {"owner","admin"}: return self.send_json(403,{"error":"forbidden"})
|
||||
return self.send_json(200,{"items":[dict(r) for r in db.execute("SELECT id,email,role,organization_id,created_at FROM users WHERE organization_id=? ORDER BY id",(org,))]})
|
||||
if path=="/api/v1/dashboard/summary":
|
||||
row=db.execute("SELECT COUNT(*) businesses,COALESCE(AVG(score),0) average_score FROM businesses WHERE organization_id=?",(org,)).fetchone(); return self.send_json(200,{"organization_id":org,"businesses":row["businesses"],"average_score":round(row["average_score"],2),"suppressed":db.execute("SELECT COUNT(*) FROM suppressions WHERE organization_id=?",(org,)).fetchone()[0]})
|
||||
row=db.execute("SELECT COUNT(*) businesses,COALESCE(AVG(score),0) average_score FROM businesses WHERE organization_id=?",(org,)).fetchone()
|
||||
counts={"new":db.execute("SELECT COUNT(*) FROM businesses WHERE organization_id=? AND created_at>=datetime('now','-7 days')",(org,)).fetchone()[0],"hot":db.execute("SELECT COUNT(*) FROM businesses WHERE organization_id=? AND score>=70 AND merge_status='active'",(org,)).fetchone()[0],"review":db.execute("SELECT COUNT(*) FROM businesses WHERE organization_id=? AND review_status='pending'",(org,)).fetchone()[0],"source_health":db.execute("SELECT COUNT(*) FROM sources WHERE organization_id=? AND health_status IN ('healthy','unhealthy')",(org,)).fetchone()[0],"active_jobs":db.execute("SELECT COUNT(*) FROM jobs WHERE organization_id=? AND status IN ('queued','running')",(org,)).fetchone()[0]}
|
||||
clickable={key:{"count":value,"filter":{"dashboard_filter":key}} for key,value in counts.items()}
|
||||
return self.send_json(200,{"organization_id":org,"businesses":row["businesses"],"average_score":round(row["average_score"],2),"suppressed":db.execute("SELECT COUNT(*) FROM suppressions WHERE organization_id=?",(org,)).fetchone()[0],"counts":counts,"clickable_filters":clickable,"quick_filters":[{"key":key,"count":value,"filter":{"dashboard_filter":key}} for key,value in counts.items()]})
|
||||
if path=="/api/v1/saved-filters": return self.list_saved_filters(db,user)
|
||||
if path=="/api/v1/review-queue": return self.review_queue(db,org,parse_qs(parsed.query))
|
||||
if path=="/api/v1/score-rules": return self.list_score_rules(db,org)
|
||||
if path=="/api/v1/scoring/summary": return self.scoring_summary(db,org)
|
||||
if path=="/api/v1/businesses": return self.list_businesses(db,org,parse_qs(parsed.query))
|
||||
@@ -456,6 +572,27 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
offset=(number("cursor",0) or 0)+(page-1)*size
|
||||
rows=db.execute("SELECT b.* FROM businesses b WHERE "+" AND ".join(where)+" ORDER BY b.score DESC,b.id LIMIT ? OFFSET ?",params+[size+1,offset]).fetchall(); more=len(rows)>size; rows=rows[:size]
|
||||
return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in rows],"page":page,"page_size":size,"next_cursor":str(offset+size) if more else None})
|
||||
def bulk_review(self, payload, db, user):
|
||||
ids = payload.get("ids", payload.get("business_ids")); action = str(payload.get("action", "")).strip().lower()
|
||||
if not isinstance(ids, list) or not ids or len(ids) > 100 or any(not isinstance(i, int) or i < 1 for i in ids) or len(set(ids)) != len(ids): return self.send_json(400, {"error": "invalid_bulk_ids"})
|
||||
if action not in {"verify", "reject", "assign"}: return self.send_json(400, {"error": "invalid_bulk_action"})
|
||||
if action == "assign":
|
||||
assignee = str(payload.get("assigned_to", payload.get("assignee", ""))).strip()
|
||||
if not assignee or len(assignee) > 120: return self.send_json(400, {"error": "assignee_required"})
|
||||
else: assignee = ""
|
||||
org = user["organization_id"]; marks = ",".join("?" for _ in ids)
|
||||
rows = db.execute("SELECT id FROM businesses WHERE organization_id=? AND id IN (" + marks + ")", [org] + ids).fetchall()
|
||||
if len(rows) != len(ids): return self.send_json(404, {"error": "not_found"})
|
||||
try:
|
||||
if action == "verify": db.execute("UPDATE businesses SET verified=1,verified_at=CURRENT_TIMESTAMP,review_status='verified',updated_at=CURRENT_TIMESTAMP WHERE organization_id=? AND id IN (" + marks + ")", [org] + ids)
|
||||
elif action == "reject": db.execute("UPDATE businesses SET verified=0,review_status='rejected',updated_at=CURRENT_TIMESTAMP WHERE organization_id=? AND id IN (" + marks + ")", [org] + ids)
|
||||
else: db.execute("UPDATE businesses SET assigned_to=?,review_status='assigned',updated_at=CURRENT_TIMESTAMP WHERE organization_id=? AND id IN (" + marks + ")", [assignee, org] + ids)
|
||||
self.audit(db, user, "businesses.bulk_review", json.dumps({"action": action, "ids": ids, "assigned_to": assignee}, sort_keys=True))
|
||||
db.commit()
|
||||
except Exception:
|
||||
db.rollback(); raise
|
||||
return self.send_json(200, {"action": action, "updated": ids, "count": len(ids)})
|
||||
|
||||
def do_POST(self):
|
||||
path=urlparse(self.path).path.rstrip("/")
|
||||
if path=="/api/v1/auth/login":return self.login(self.read_json())
|
||||
@@ -472,6 +609,8 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
return self.create_job(self.read_json(),db,user)
|
||||
if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"})
|
||||
payload=self.read_json(); org=user["organization_id"]
|
||||
if path=="/api/v1/saved-filters": return self.save_filter(payload,db,user)
|
||||
if path=="/api/v1/businesses/bulk-review": return self.bulk_review(payload,db,user)
|
||||
if path=="/api/v1/score-rules": return self.create_score_rule(payload,db,user)
|
||||
if len(path.split("/"))==7 and path.split("/")[3:]==["businesses",path.split("/")[4],"score","recalculate"]: return self.recalculate_score(int(path.split("/")[4]) if path.split("/")[4].isdigit() else -1,payload,db,user)
|
||||
if path.startswith("/api/v1/jobs/"):
|
||||
@@ -506,11 +645,23 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
if not user:return
|
||||
if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"})
|
||||
bits=path.split("/")
|
||||
if len(bits)==5 and bits[:4]==["","api","v1","saved-filters"] and bits[4].isdigit(): return self.update_saved_filter(int(bits[4]),self.read_json(),db,user)
|
||||
if len(bits)==5 and bits[:4]==["","api","v1","score-rules"] and bits[4].isdigit(): return self.update_score_rule(int(bits[4]),self.read_json(),db,user)
|
||||
if len(bits)==5 and bits[:4]==["","api","v1","sources"] and bits[4].isdigit(): return self.update_source(int(bits[4]),self.read_json(),db,user)
|
||||
if len(bits)==6 and bits[:4]==["","api","v1","businesses"] and bits[5]=="pipeline":return self.update_pipeline(int(bits[4]) if bits[4].isdigit() else -1,self.read_json(),db,user)
|
||||
return self.send_json(404,{"error":"not_found"})
|
||||
finally:db.close()
|
||||
def do_DELETE(self):
|
||||
path=urlparse(self.path).path.rstrip("/"); db=self.db()
|
||||
try:
|
||||
user=self.require_auth(db)
|
||||
if not user:return
|
||||
if user["role"] not in MUTATING_ROLES:return self.send_json(403,{"error":"forbidden"})
|
||||
bits=path.split("/")
|
||||
if len(bits)==5 and bits[:4]==["","api","v1","saved-filters"] and bits[4].isdigit(): return self.delete_saved_filter(int(bits[4]),db,user)
|
||||
return self.send_json(404,{"error":"not_found"})
|
||||
finally: db.close()
|
||||
|
||||
def login(self,payload):
|
||||
db=self.db(); email=str(payload.get("email"," ")).strip().lower(); password=str(payload.get("password","")); user=db.execute("SELECT * FROM users WHERE email=?",(email,)).fetchone()
|
||||
try:
|
||||
@@ -558,7 +709,7 @@ class ApiHandler(BaseHTTPRequestHandler):
|
||||
stage=str(payload["stage"]).strip();status=str(payload.get("status","active")).strip() or "active";db.execute("INSERT INTO pipeline_entries(business_id,organization_id,stage,status) VALUES(?,?,?,?)",(bid,user["organization_id"],stage,status));rid=db.execute("SELECT last_insert_rowid()").fetchone()[0];self.audit(db,user,"pipeline.updated",stage);db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM pipeline_entries WHERE id=?",(rid,)).fetchone()))
|
||||
def verify_business(self,bid,payload,db,user):
|
||||
if not self.child_business(db,bid,user):return self.send_json(404,{"error":"not_found"})
|
||||
verified=bool(payload.get("verified",True)); now=datetime.now(timezone.utc).replace(microsecond=0).isoformat();db.execute("UPDATE businesses SET verified=?,verified_at=?,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(int(verified),now if verified else None,bid,user["organization_id"]));self.audit(db,user,"business.verified",str(verified));db.commit();row=self.business(db,bid,user["organization_id"]);return self.send_json(200,row_json(row))
|
||||
verified=bool(payload.get("verified",True)); now=datetime.now(timezone.utc).replace(microsecond=0).isoformat();db.execute("UPDATE businesses SET verified=?,verified_at=?,review_status=?,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(int(verified),now if verified else None,"verified" if verified else "pending",bid,user["organization_id"]));self.audit(db,user,"business.verified",str(verified));db.commit();row=self.business(db,bid,user["organization_id"]);return self.send_json(200,row_json(row))
|
||||
def preview_import(self,payload,db,org):
|
||||
rows=payload.get("rows",[])
|
||||
if not isinstance(rows,list):return self.send_json(400,{"error":"rows_required"})
|
||||
|
||||
Reference in New Issue
Block a user