255 lines
8.7 KiB
Python
255 lines
8.7 KiB
Python
"""Executes a report definition against the analytics read layer.
|
|
|
|
A report is a saved *parameterisation* of the same governed queries the
|
|
dashboard runs — never free-form SQL. That keeps ADR-0010's constraint intact:
|
|
adding a report type means adding a builder here, not opening a query surface.
|
|
|
|
Every builder returns the same tabular envelope so one DataTable and one CSV
|
|
writer can render any report type:
|
|
|
|
{"columns": [{"key", "label"}, ...], "rows": [dict, ...]}
|
|
|
|
Filters accept either explicit ISO `from_date`/`to_date` bounds or a rolling
|
|
`window_days`, resolved at run time. Rolling is the default a saved report
|
|
wants — "last 90 days" should mean the last 90 days on every run, not the
|
|
quarter that was current when the report was saved.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from analytics.views import Analytics
|
|
from job.job_post.models import JobPosts
|
|
|
|
REPORT_TYPES = (
|
|
"kpis",
|
|
"funnel",
|
|
"hiring_trend",
|
|
"source_performance",
|
|
"recruiter_performance",
|
|
"department_performance",
|
|
)
|
|
|
|
REPORT_TYPE_LABELS = {
|
|
"kpis": "KPI Summary",
|
|
"funnel": "Hiring Funnel",
|
|
"hiring_trend": "Hiring Trend",
|
|
"source_performance": "Source Performance",
|
|
"recruiter_performance": "Recruiter Performance",
|
|
"department_performance": "Department Performance",
|
|
}
|
|
|
|
# Keys a saved filter object may carry; anything else is dropped on save.
|
|
FILTER_KEYS = ("from_date", "to_date", "window_days", "department", "recruiter_id", "months", "top")
|
|
|
|
DEPT_CAP = 24
|
|
|
|
_KPI_ROWS = (
|
|
("open_jobs", "Open Jobs"),
|
|
("closed_jobs", "Closed Jobs"),
|
|
("total_candidates", "Total Candidates"),
|
|
("hires", "Hires"),
|
|
("offers_sent", "Offers Sent"),
|
|
("offers_accepted", "Offers Accepted"),
|
|
("time_to_hire", "Time to Hire (days)"),
|
|
("time_to_fill", "Time to Fill (days)"),
|
|
("cost_per_hire", "Cost per Hire"),
|
|
)
|
|
|
|
|
|
def _parse_dt(value):
|
|
if value in (None, ""):
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def resolve_filters(filters: dict | None) -> dict:
|
|
"""Normalize a saved/ad-hoc filter object into keyword args for Analytics."""
|
|
filters = filters if isinstance(filters, dict) else {}
|
|
from_date = _parse_dt(filters.get("from_date"))
|
|
to_date = _parse_dt(filters.get("to_date"))
|
|
window_days = filters.get("window_days")
|
|
if from_date is None and to_date is None and window_days:
|
|
try:
|
|
days = max(1, min(int(window_days), 3650))
|
|
except (TypeError, ValueError):
|
|
days = None
|
|
if days:
|
|
to_date = datetime.now(timezone.utc)
|
|
from_date = to_date - timedelta(days=days)
|
|
department = (filters.get("department") or "").strip() or None
|
|
recruiter_id = (filters.get("recruiter_id") or "").strip() or None
|
|
try:
|
|
months = max(1, min(int(filters.get("months") or 7), 24))
|
|
except (TypeError, ValueError):
|
|
months = 7
|
|
try:
|
|
top = max(1, min(int(filters.get("top") or 10), 50))
|
|
except (TypeError, ValueError):
|
|
top = 10
|
|
return {
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"department": department,
|
|
"recruiter_id": recruiter_id,
|
|
"months": months,
|
|
"top": top,
|
|
}
|
|
|
|
|
|
def _round(value, digits=1):
|
|
if value is None:
|
|
return None
|
|
value = round(float(value), digits)
|
|
# Integral values export as "3", not "3.0" — counts are ints in the CSV.
|
|
return int(value) if value.is_integer() else value
|
|
|
|
|
|
async def _build_kpis(session, f):
|
|
service = Analytics(session=session)
|
|
data = await service.get_kpis(f["from_date"], f["to_date"], f["department"], f["recruiter_id"])
|
|
rows = []
|
|
for key, label in _KPI_ROWS:
|
|
rows.append({
|
|
"metric": label,
|
|
"current": _round(data.get(key)),
|
|
"prior": _round(data.get(f"{key}_prior")),
|
|
})
|
|
columns = [
|
|
{"key": "metric", "label": "Metric"},
|
|
{"key": "current", "label": "Current Window"},
|
|
{"key": "prior", "label": "Prior Window"},
|
|
]
|
|
return columns, rows
|
|
|
|
|
|
async def _build_funnel(session, f):
|
|
service = Analytics(session=session)
|
|
data = await service.get_funnel(f["from_date"], f["to_date"], f["department"], f["recruiter_id"])
|
|
columns = [
|
|
{"key": "stage", "label": "Stage"},
|
|
{"key": "count", "label": "Applications"},
|
|
]
|
|
return columns, list(data)
|
|
|
|
|
|
async def _build_hiring_trend(session, f):
|
|
service = Analytics(session=session)
|
|
data = await service.get_hiring_trend(
|
|
f["months"], f["from_date"], f["to_date"], f["department"], f["recruiter_id"]
|
|
)
|
|
labels = data.get("labels") or []
|
|
apps = data.get("applications") or []
|
|
hires = data.get("hires") or []
|
|
rows = [
|
|
{"month": labels[i], "applications": apps[i], "hires": hires[i]}
|
|
for i in range(len(labels))
|
|
]
|
|
columns = [
|
|
{"key": "month", "label": "Month"},
|
|
{"key": "applications", "label": "Applications"},
|
|
{"key": "hires", "label": "Hires"},
|
|
]
|
|
return columns, rows
|
|
|
|
|
|
async def _build_source_performance(session, f):
|
|
service = Analytics(session=session)
|
|
rows = await service.get_source_performance(
|
|
f["from_date"], f["to_date"], f["department"], f["recruiter_id"]
|
|
)
|
|
columns = [
|
|
{"key": "source", "label": "Source"},
|
|
{"key": "count", "label": "Applications"},
|
|
{"key": "spend", "label": "Spend"},
|
|
{"key": "cost_per_application", "label": "Cost per Application"},
|
|
]
|
|
return columns, list(rows)
|
|
|
|
|
|
async def _build_recruiter_performance(session, f):
|
|
service = Analytics(session=session)
|
|
rows = await service.get_recruiter_performance(
|
|
f["top"], f["from_date"], f["to_date"], f["department"], f["recruiter_id"]
|
|
)
|
|
for row in rows:
|
|
row["avg_time_to_hire"] = _round(row.get("avg_time_to_hire"))
|
|
columns = [
|
|
{"key": "name", "label": "Recruiter"},
|
|
{"key": "completed", "label": "Completed Requisitions"},
|
|
{"key": "hires", "label": "Hires"},
|
|
{"key": "open_reqs", "label": "Open Requisitions"},
|
|
{"key": "avg_time_to_hire", "label": "Avg Time to Hire (days)"},
|
|
]
|
|
return columns, rows
|
|
|
|
|
|
async def _build_department_performance(session, f):
|
|
statement = (
|
|
select(JobPosts.department)
|
|
.where(JobPosts.is_deleted == False, JobPosts.department.is_not(None)) # noqa: E712
|
|
.distinct()
|
|
.order_by(JobPosts.department.asc())
|
|
.limit(DEPT_CAP)
|
|
)
|
|
departments = [d for (d,) in (await session.execute(statement)).all() if d]
|
|
service = Analytics(session=session)
|
|
rows = []
|
|
for dept in departments:
|
|
data = await service.get_kpis(f["from_date"], f["to_date"], dept, f["recruiter_id"])
|
|
rows.append({
|
|
"department": dept,
|
|
"open_jobs": data.get("open_jobs") or 0,
|
|
"applications": data.get("total_candidates") or 0,
|
|
"hires": data.get("hires") or 0,
|
|
"time_to_fill": _round(data.get("time_to_fill")),
|
|
})
|
|
rows = [r for r in rows if r["open_jobs"] or r["applications"] or r["hires"]]
|
|
rows.sort(key=lambda r: r["hires"], reverse=True)
|
|
columns = [
|
|
{"key": "department", "label": "Department"},
|
|
{"key": "open_jobs", "label": "Open Roles"},
|
|
{"key": "applications", "label": "Applications"},
|
|
{"key": "hires", "label": "Hires"},
|
|
{"key": "time_to_fill", "label": "Time to Fill (days)"},
|
|
]
|
|
return columns, rows
|
|
|
|
|
|
_BUILDERS = {
|
|
"kpis": _build_kpis,
|
|
"funnel": _build_funnel,
|
|
"hiring_trend": _build_hiring_trend,
|
|
"source_performance": _build_source_performance,
|
|
"recruiter_performance": _build_recruiter_performance,
|
|
"department_performance": _build_department_performance,
|
|
}
|
|
|
|
|
|
async def run_report(session: AsyncSession, report_type: str, filters: dict | None) -> dict:
|
|
builder = _BUILDERS.get(report_type)
|
|
if builder is None:
|
|
raise ValueError(f"unknown report type: {report_type}")
|
|
resolved = resolve_filters(filters)
|
|
columns, rows = await builder(session, resolved)
|
|
return {
|
|
"report_type": report_type,
|
|
"report_label": REPORT_TYPE_LABELS.get(report_type, report_type),
|
|
"columns": columns,
|
|
"rows": rows,
|
|
"row_count": len(rows),
|
|
"window": {
|
|
"from_date": resolved["from_date"].isoformat() if resolved["from_date"] else None,
|
|
"to_date": resolved["to_date"].isoformat() if resolved["to_date"] else None,
|
|
},
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|