425 lines
16 KiB
Python
425 lines
16 KiB
Python
|
"""Revenue Cockpit — DB operations, import, tracking, agent tasks."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from datetime import date, datetime, timezone
|
||
|
|
from typing import Any, Optional
|
||
|
|
|
||
|
|
from app.db import execute, fetch_all, fetch_one
|
||
|
|
from app.services import agent_integration
|
||
|
|
from app.services.excel_import import DEFAULT_FILE, DEFAULT_SHEET, parse_revenue_sheet
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_excel_parse(path: str = DEFAULT_FILE, sheet: str | None = DEFAULT_SHEET) -> dict[str, Any]:
|
||
|
|
result = parse_revenue_sheet(path, sheet)
|
||
|
|
if not result.get("ok"):
|
||
|
|
raise RuntimeError(result.get("error") or "Excel parse failed")
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
async def fetch_excel_parse_async(path: str = DEFAULT_FILE, sheet: str | None = DEFAULT_SHEET) -> dict[str, Any]:
|
||
|
|
return fetch_excel_parse(path, sheet)
|
||
|
|
|
||
|
|
|
||
|
|
def _ser(row: dict[str, Any] | None) -> dict[str, Any] | None:
|
||
|
|
if not row:
|
||
|
|
return None
|
||
|
|
out = dict(row)
|
||
|
|
for k, v in list(out.items()):
|
||
|
|
if hasattr(v, "isoformat"):
|
||
|
|
out[k] = v.isoformat()
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def _log_change(entity_type: str, entity_id: int, field: str, old: Any, new: Any, by: str = "ceo") -> None:
|
||
|
|
execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_change_log (entity_type, entity_id, field_name, old_value, new_value, changed_by)
|
||
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
||
|
|
""",
|
||
|
|
(entity_type, entity_id, field, str(old) if old is not None else None, str(new) if new is not None else None, by),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def import_from_parsed(parsed: dict[str, Any], imported_by: str = "ceo", replace: bool = True) -> dict[str, Any]:
|
||
|
|
if replace:
|
||
|
|
execute("DELETE FROM revenue_objectives WHERE project_id IN (SELECT id FROM revenue_projects)")
|
||
|
|
execute("DELETE FROM revenue_projects")
|
||
|
|
execute("UPDATE revenue_cockpit_goals SET is_active = FALSE WHERE is_active = TRUE")
|
||
|
|
|
||
|
|
goals = parsed.get("goals") or {}
|
||
|
|
g = fetch_one(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_cockpit_goals (vision_text, horizon_text, mid_text, tagline, is_active)
|
||
|
|
VALUES (%s, %s, %s, %s, TRUE)
|
||
|
|
RETURNING *
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
goals.get("vision_text", ""),
|
||
|
|
goals.get("horizon_text", ""),
|
||
|
|
goals.get("mid_text", ""),
|
||
|
|
goals.get("tagline", ""),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
count = 0
|
||
|
|
for p in parsed.get("projects") or []:
|
||
|
|
row = fetch_one(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_projects
|
||
|
|
(name, category, margin_month, margin_year, target_revenue, next_steps, status, sort_order, metadata)
|
||
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb)
|
||
|
|
RETURNING id
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
p["name"],
|
||
|
|
p.get("category", "deal"),
|
||
|
|
p.get("margin_month"),
|
||
|
|
p.get("margin_year"),
|
||
|
|
p.get("target_revenue"),
|
||
|
|
p.get("next_steps"),
|
||
|
|
p.get("status", "active"),
|
||
|
|
p.get("sort_order", count),
|
||
|
|
json.dumps({
|
||
|
|
"source_row": p.get("source_row"),
|
||
|
|
"row_style": p.get("row_style", "white"),
|
||
|
|
"imported": True,
|
||
|
|
}),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
pid = row["id"] if row else None
|
||
|
|
if pid and p.get("next_steps"):
|
||
|
|
for i, line in enumerate([ln.strip() for ln in p["next_steps"].split("\n") if ln.strip()][:5]):
|
||
|
|
execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_objectives (project_id, title, description, status, sort_order)
|
||
|
|
VALUES (%s, %s, %s, 'open', %s)
|
||
|
|
""",
|
||
|
|
(pid, line[:255], p["next_steps"] if i == 0 else None, i),
|
||
|
|
)
|
||
|
|
count += 1
|
||
|
|
|
||
|
|
run = fetch_one(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_import_runs (source_file, sheet_name, rows_imported, goals_imported, imported_by, metadata)
|
||
|
|
VALUES (%s, %s, %s, TRUE, %s, %s::jsonb)
|
||
|
|
RETURNING *
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
parsed.get("source_file", ""),
|
||
|
|
parsed.get("sheet_name", ""),
|
||
|
|
count,
|
||
|
|
imported_by,
|
||
|
|
json.dumps({"parsed_at": parsed.get("parsed_at"), "file_mtime": parsed.get("file_mtime")}),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
take_snapshot()
|
||
|
|
return {"ok": True, "projects_imported": count, "goals": _ser(g), "import_run": _ser(run)}
|
||
|
|
|
||
|
|
|
||
|
|
def get_active_goals() -> dict[str, Any] | None:
|
||
|
|
return _ser(fetch_one("SELECT * FROM revenue_cockpit_goals WHERE is_active = TRUE ORDER BY id DESC LIMIT 1"))
|
||
|
|
|
||
|
|
|
||
|
|
def update_goals(data: dict[str, Any], changed_by: str = "ceo") -> dict[str, Any] | None:
|
||
|
|
old = fetch_one("SELECT * FROM revenue_cockpit_goals WHERE is_active = TRUE ORDER BY id DESC LIMIT 1")
|
||
|
|
if not old:
|
||
|
|
return None
|
||
|
|
fields = ("vision_text", "horizon_text", "mid_text", "tagline")
|
||
|
|
sets, params = [], []
|
||
|
|
for k in fields:
|
||
|
|
if k in data:
|
||
|
|
sets.append(f"{k} = %s")
|
||
|
|
params.append(data[k])
|
||
|
|
if str(old.get(k)) != str(data[k]):
|
||
|
|
_log_change("goals", old["id"], k, old.get(k), data[k], changed_by)
|
||
|
|
if not sets:
|
||
|
|
return get_active_goals()
|
||
|
|
params.append(old["id"])
|
||
|
|
execute(
|
||
|
|
f"UPDATE revenue_cockpit_goals SET {', '.join(sets)}, updated_at = NOW() WHERE id = %s",
|
||
|
|
tuple(params),
|
||
|
|
)
|
||
|
|
return get_active_goals()
|
||
|
|
|
||
|
|
|
||
|
|
def list_projects(status: str | None = None) -> list[dict[str, Any]]:
|
||
|
|
clauses, params = [], []
|
||
|
|
if status:
|
||
|
|
clauses.append("status = %s")
|
||
|
|
params.append(status)
|
||
|
|
where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
|
||
|
|
rows = fetch_all(
|
||
|
|
f"""
|
||
|
|
SELECT p.*,
|
||
|
|
(SELECT COUNT(*) FROM revenue_objectives o WHERE o.project_id = p.id AND o.status = 'open') AS open_objectives,
|
||
|
|
(SELECT COUNT(*) FROM agent_tasks t WHERE t.revenue_project_id = p.id AND t.status NOT IN ('completed','cancelled')) AS open_tasks
|
||
|
|
FROM revenue_projects p
|
||
|
|
{where}
|
||
|
|
ORDER BY p.sort_order ASC, p.name ASC
|
||
|
|
""",
|
||
|
|
tuple(params) if params else None,
|
||
|
|
)
|
||
|
|
return [_ser({**r, "row_style": _infer_row_style(r.get("name"), r.get("margin_month"), r.get("metadata"))}) for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def get_project(project_id: int) -> dict[str, Any] | None:
|
||
|
|
p = _ser(fetch_one("SELECT * FROM revenue_projects WHERE id = %s", (project_id,)))
|
||
|
|
if not p:
|
||
|
|
return None
|
||
|
|
p["objectives"] = [_ser(r) for r in fetch_all(
|
||
|
|
"SELECT * FROM revenue_objectives WHERE project_id = %s ORDER BY sort_order, id",
|
||
|
|
(project_id,),
|
||
|
|
)]
|
||
|
|
p["tasks"] = [_ser(r) for r in fetch_all(
|
||
|
|
"""
|
||
|
|
SELECT * FROM agent_tasks WHERE revenue_project_id = %s
|
||
|
|
ORDER BY created_at DESC LIMIT 50
|
||
|
|
""",
|
||
|
|
(project_id,),
|
||
|
|
)]
|
||
|
|
p["history"] = [_ser(r) for r in fetch_all(
|
||
|
|
"""
|
||
|
|
SELECT * FROM revenue_change_log
|
||
|
|
WHERE (entity_type = 'project' AND entity_id = %s)
|
||
|
|
OR (entity_type = 'objective' AND entity_id IN (
|
||
|
|
SELECT id FROM revenue_objectives WHERE project_id = %s))
|
||
|
|
ORDER BY created_at DESC LIMIT 30
|
||
|
|
""",
|
||
|
|
(project_id, project_id),
|
||
|
|
)]
|
||
|
|
p["row_style"] = _infer_row_style(p.get("name"), p.get("margin_month"), p.get("metadata"))
|
||
|
|
return p
|
||
|
|
|
||
|
|
|
||
|
|
def set_project_row_style(project_id: int, row_style: str) -> None:
|
||
|
|
old = fetch_one("SELECT metadata FROM revenue_projects WHERE id = %s", (project_id,))
|
||
|
|
meta = dict(old.get("metadata") or {}) if old else {}
|
||
|
|
meta["row_style"] = row_style
|
||
|
|
execute(
|
||
|
|
"UPDATE revenue_projects SET metadata = %s::jsonb, updated_at = NOW() WHERE id = %s",
|
||
|
|
(json.dumps(meta), project_id),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def update_project(project_id: int, data: dict[str, Any], changed_by: str = "ceo") -> dict[str, Any] | None:
|
||
|
|
old = fetch_one("SELECT * FROM revenue_projects WHERE id = %s", (project_id,))
|
||
|
|
if not old:
|
||
|
|
return None
|
||
|
|
fields = {
|
||
|
|
"name": data.get("name"),
|
||
|
|
"category": data.get("category"),
|
||
|
|
"margin_month": data.get("margin_month"),
|
||
|
|
"margin_year": data.get("margin_year"),
|
||
|
|
"target_revenue": data.get("target_revenue"),
|
||
|
|
"next_steps": data.get("next_steps"),
|
||
|
|
"status": data.get("status"),
|
||
|
|
"priority": data.get("priority"),
|
||
|
|
}
|
||
|
|
sets, params = [], []
|
||
|
|
for k, v in fields.items():
|
||
|
|
if k in data:
|
||
|
|
sets.append(f"{k} = %s")
|
||
|
|
params.append(v)
|
||
|
|
if str(old.get(k)) != str(v):
|
||
|
|
_log_change("project", project_id, k, old.get(k), v, changed_by)
|
||
|
|
if not sets:
|
||
|
|
return get_project(project_id)
|
||
|
|
params.append(project_id)
|
||
|
|
execute(f"UPDATE revenue_projects SET {', '.join(sets)}, updated_at = NOW() WHERE id = %s", tuple(params))
|
||
|
|
return get_project(project_id)
|
||
|
|
|
||
|
|
|
||
|
|
def _infer_row_style(name: str, margin_month: Any, metadata: Any) -> str:
|
||
|
|
if isinstance(metadata, dict) and metadata.get("row_style"):
|
||
|
|
return str(metadata["row_style"])
|
||
|
|
n = (name or "").lower()
|
||
|
|
if "foodlinkk food marketing" in n or "total earnings" in n or "loonkosten aissa" in n:
|
||
|
|
return "yellow"
|
||
|
|
if "foodservice" in n or n.strip() == "dirk":
|
||
|
|
return "orange"
|
||
|
|
if margin_month is not None and margin_month > 0:
|
||
|
|
return "green"
|
||
|
|
if any(x in n for x in ("plus", "vomar", "hoogvliet", "deka", "spar", "doner palace", "zakat")):
|
||
|
|
return "red"
|
||
|
|
return "white"
|
||
|
|
|
||
|
|
|
||
|
|
def create_objective(project_id: int, title: str, description: str | None = None, priority: str = "normal") -> dict[str, Any]:
|
||
|
|
row = fetch_one(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_objectives (project_id, title, description, priority)
|
||
|
|
VALUES (%s, %s, %s, %s)
|
||
|
|
RETURNING *
|
||
|
|
""",
|
||
|
|
(project_id, title.strip(), description, priority),
|
||
|
|
)
|
||
|
|
_log_change("objective", row["id"], "created", None, title, "ceo")
|
||
|
|
return _ser(row) or {}
|
||
|
|
|
||
|
|
|
||
|
|
def update_objective(objective_id: int, data: dict[str, Any]) -> dict[str, Any] | None:
|
||
|
|
old = fetch_one("SELECT * FROM revenue_objectives WHERE id = %s", (objective_id,))
|
||
|
|
if not old:
|
||
|
|
return None
|
||
|
|
for k in ("title", "description", "status", "priority", "due_date"):
|
||
|
|
if k in data:
|
||
|
|
execute(
|
||
|
|
f"UPDATE revenue_objectives SET {k} = %s, updated_at = NOW() WHERE id = %s",
|
||
|
|
(data[k], objective_id),
|
||
|
|
)
|
||
|
|
_log_change("objective", objective_id, k, old.get(k), data[k])
|
||
|
|
return _ser(fetch_one("SELECT * FROM revenue_objectives WHERE id = %s", (objective_id,)))
|
||
|
|
|
||
|
|
|
||
|
|
def dashboard_stats() -> dict[str, Any]:
|
||
|
|
goals = get_active_goals()
|
||
|
|
totals = fetch_one(
|
||
|
|
"""
|
||
|
|
SELECT
|
||
|
|
COUNT(*) FILTER (WHERE status = 'active') AS active_projects,
|
||
|
|
COALESCE(SUM(margin_year) FILTER (WHERE status = 'active'), 0) AS total_margin_year,
|
||
|
|
COALESCE(SUM(margin_month) FILTER (WHERE status = 'active'), 0) AS total_margin_month,
|
||
|
|
COUNT(*) FILTER (WHERE category = 'deal' AND status = 'active') AS active_deals
|
||
|
|
FROM revenue_projects
|
||
|
|
"""
|
||
|
|
) or {}
|
||
|
|
open_obj = fetch_one("SELECT COUNT(*) AS n FROM revenue_objectives WHERE status = 'open'") or {"n": 0}
|
||
|
|
open_tasks = fetch_one(
|
||
|
|
"SELECT COUNT(*) AS n FROM agent_tasks WHERE status NOT IN ('completed','cancelled') AND revenue_project_id IS NOT NULL"
|
||
|
|
) or {"n": 0}
|
||
|
|
prev = fetch_one(
|
||
|
|
"SELECT * FROM revenue_snapshots WHERE snapshot_date < CURRENT_DATE ORDER BY snapshot_date DESC LIMIT 1"
|
||
|
|
)
|
||
|
|
return {
|
||
|
|
"goals": goals,
|
||
|
|
"active_projects": int(totals.get("active_projects") or 0),
|
||
|
|
"active_deals": int(totals.get("active_deals") or 0),
|
||
|
|
"total_margin_year": float(totals.get("total_margin_year") or 0),
|
||
|
|
"total_margin_month": float(totals.get("total_margin_month") or 0),
|
||
|
|
"open_objectives": int(open_obj.get("n") or 0),
|
||
|
|
"open_agent_tasks": int(open_tasks.get("n") or 0),
|
||
|
|
"previous_snapshot": _ser(prev),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def take_snapshot() -> dict[str, Any]:
|
||
|
|
stats = dashboard_stats()
|
||
|
|
row = fetch_one(
|
||
|
|
"""
|
||
|
|
INSERT INTO revenue_snapshots
|
||
|
|
(snapshot_date, total_margin_year, total_margin_month, active_projects, open_objectives, open_agent_tasks, payload)
|
||
|
|
VALUES (CURRENT_DATE, %s, %s, %s, %s, %s, %s::jsonb)
|
||
|
|
ON CONFLICT (snapshot_date) DO UPDATE SET
|
||
|
|
total_margin_year = EXCLUDED.total_margin_year,
|
||
|
|
total_margin_month = EXCLUDED.total_margin_month,
|
||
|
|
active_projects = EXCLUDED.active_projects,
|
||
|
|
open_objectives = EXCLUDED.open_objectives,
|
||
|
|
open_agent_tasks = EXCLUDED.open_agent_tasks,
|
||
|
|
payload = EXCLUDED.payload,
|
||
|
|
created_at = NOW()
|
||
|
|
RETURNING *
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
stats["total_margin_year"],
|
||
|
|
stats["total_margin_month"],
|
||
|
|
stats["active_projects"],
|
||
|
|
stats["open_objectives"],
|
||
|
|
stats["open_agent_tasks"],
|
||
|
|
json.dumps({"goals_id": (stats.get("goals") or {}).get("id")}),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
return _ser(row) or {}
|
||
|
|
|
||
|
|
|
||
|
|
def list_snapshots(limit: int = 90) -> list[dict[str, Any]]:
|
||
|
|
rows = fetch_all(
|
||
|
|
"SELECT * FROM revenue_snapshots ORDER BY snapshot_date DESC LIMIT %s",
|
||
|
|
(max(1, min(limit, 365)),),
|
||
|
|
)
|
||
|
|
return [_ser(r) for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def assign_agent_task(
|
||
|
|
project_id: int,
|
||
|
|
agent_name: str,
|
||
|
|
title: str,
|
||
|
|
description: str | None = None,
|
||
|
|
objective_id: int | None = None,
|
||
|
|
priority: str = "normal",
|
||
|
|
delegate_herman: bool = True,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
project = fetch_one("SELECT name FROM revenue_projects WHERE id = %s", (project_id,))
|
||
|
|
if not project:
|
||
|
|
raise ValueError("Project not found")
|
||
|
|
agent = agent_name.strip().lower()
|
||
|
|
desc = description or ""
|
||
|
|
row = fetch_one(
|
||
|
|
"""
|
||
|
|
INSERT INTO agent_tasks
|
||
|
|
(agent_name, title, description, status, priority, assigned_by, revenue_project_id, revenue_objective_id, source)
|
||
|
|
VALUES (%s, %s, %s, 'pending', %s, 'ceo', %s, %s, 'revenue_cockpit')
|
||
|
|
RETURNING *
|
||
|
|
""",
|
||
|
|
(agent, title.strip(), desc, priority, project_id, objective_id),
|
||
|
|
)
|
||
|
|
task = _ser(row) or {}
|
||
|
|
body = f"Revenue Cockpit · {project['name']}: {title}"
|
||
|
|
if desc:
|
||
|
|
body += f"\n{desc}"
|
||
|
|
try:
|
||
|
|
execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO agent_events (agent_name, agent_type, event_type, title, body, status, channel, metadata, related_table, related_id)
|
||
|
|
VALUES (%s, 'revenue_cockpit', 'task_assigned', %s, %s, 'pending', 'revenue_cockpit', %s::jsonb, 'agent_tasks', %s)
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
agent,
|
||
|
|
title.strip(),
|
||
|
|
body,
|
||
|
|
json.dumps({"project_id": project_id, "objective_id": objective_id, "task_id": task.get("id")}),
|
||
|
|
task.get("id"),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
if delegate_herman and agent != "herman":
|
||
|
|
try:
|
||
|
|
agent_integration.create_handoff(
|
||
|
|
"ceo",
|
||
|
|
agent,
|
||
|
|
handoff_type="task",
|
||
|
|
payload={"task_id": task.get("id"), "project_id": project_id, "title": title},
|
||
|
|
status="pending",
|
||
|
|
)
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
_log_change("project", project_id, "agent_task", None, title, "ceo")
|
||
|
|
return task
|
||
|
|
|
||
|
|
|
||
|
|
def list_agent_tasks(limit: int = 50) -> list[dict[str, Any]]:
|
||
|
|
rows = fetch_all(
|
||
|
|
"""
|
||
|
|
SELECT t.*, p.name AS project_name
|
||
|
|
FROM agent_tasks t
|
||
|
|
LEFT JOIN revenue_projects p ON p.id = t.revenue_project_id
|
||
|
|
WHERE t.revenue_project_id IS NOT NULL
|
||
|
|
ORDER BY t.created_at DESC
|
||
|
|
LIMIT %s
|
||
|
|
""",
|
||
|
|
(max(1, min(limit, 200)),),
|
||
|
|
)
|
||
|
|
return [_ser(r) for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
async def delegate_via_herman(project_id: int, message: str) -> dict[str, Any]:
|
||
|
|
from app.services import herman as herman_service
|
||
|
|
|
||
|
|
project = get_project(project_id)
|
||
|
|
if not project:
|
||
|
|
raise ValueError("Project not found")
|
||
|
|
prompt = f"[Revenue Cockpit · {project['name']}] {message}"
|
||
|
|
result = await herman_service.chat(prompt)
|
||
|
|
return {"ok": True, "result": result, "project_id": project_id}
|