145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
from .routes_shared import * # noqa: F401,F403
|
|
from datetime import date, timedelta
|
|
from .routes_reporting_api import build_report_columns_meta, build_report_job_filters_meta
|
|
from sqlalchemy import cast, String
|
|
|
|
def get_default_report_period():
|
|
"""Return default report period (last 7 days)."""
|
|
period_end = date.today()
|
|
period_start = period_end - timedelta(days=7)
|
|
return period_start, period_end
|
|
|
|
|
|
|
|
def _safe_json_list(value):
|
|
if not value:
|
|
return []
|
|
try:
|
|
if isinstance(value, (list, tuple)):
|
|
return [int(v) for v in value]
|
|
return json.loads(value)
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _safe_json_dict(value):
|
|
if not value:
|
|
return {}
|
|
if isinstance(value, dict):
|
|
return value
|
|
try:
|
|
return json.loads(value)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _build_report_item(r):
|
|
return {
|
|
"id": int(r.id),
|
|
"name": r.name or "",
|
|
"description": r.description or "",
|
|
"report_type": r.report_type,
|
|
"output_format": r.output_format,
|
|
"customer_scope": getattr(r, "customer_scope", "all") or "all",
|
|
"customer_ids": _safe_json_list(getattr(r, "customer_ids", None)),
|
|
"period_start": r.period_start.isoformat() if getattr(r, "period_start", None) else "",
|
|
"period_end": r.period_end.isoformat() if getattr(r, "period_end", None) else "",
|
|
"schedule": r.schedule or "",
|
|
"report_config": _safe_json_dict(getattr(r, "report_config", None)),
|
|
"created_at": r.created_at.isoformat() if getattr(r, "created_at", None) else "",
|
|
}
|
|
|
|
|
|
@main_bp.route("/reports")
|
|
@login_required
|
|
def reports():
|
|
q = (request.args.get("q") or "").strip()
|
|
|
|
def _patterns(raw: str) -> list[str]:
|
|
out = []
|
|
for tok in [t.strip() for t in (raw or "").split() if t.strip()]:
|
|
p = tok.replace("\\", "\\\\")
|
|
p = p.replace("%", "\\%").replace("_", "\\_")
|
|
p = p.replace("*", "%")
|
|
if not p.startswith("%"):
|
|
p = "%" + p
|
|
if not p.endswith("%"):
|
|
p = p + "%"
|
|
out.append(p)
|
|
return out
|
|
|
|
# Pre-render items so the page is usable even if JS fails to load/execute.
|
|
query = db.session.query(ReportDefinition)
|
|
if q:
|
|
for pat in _patterns(q):
|
|
query = query.filter(
|
|
(func.coalesce(ReportDefinition.name, "").ilike(pat, escape="\\"))
|
|
| (func.coalesce(ReportDefinition.report_type, "").ilike(pat, escape="\\"))
|
|
| (func.coalesce(ReportDefinition.output_format, "").ilike(pat, escape="\\"))
|
|
| (cast(ReportDefinition.period_start, String).ilike(pat, escape="\\"))
|
|
| (cast(ReportDefinition.period_end, String).ilike(pat, escape="\\"))
|
|
)
|
|
rows = query.order_by(ReportDefinition.created_at.desc()).limit(200).all()
|
|
items = [_build_report_item(r) for r in rows]
|
|
|
|
period_start, period_end = get_default_report_period()
|
|
|
|
return render_template(
|
|
"main/reports.html",
|
|
initial_reports=items,
|
|
columns_meta=build_report_columns_meta(),
|
|
job_filters_meta=build_report_job_filters_meta(),
|
|
default_period_start=period_start.isoformat(),
|
|
default_period_end=period_end.isoformat(),
|
|
q=q,
|
|
)
|
|
|
|
|
|
@main_bp.route("/reports/new")
|
|
@login_required
|
|
def reports_new():
|
|
# Preload customers so the form remains usable if JS fails to load/execute.
|
|
customers = (
|
|
db.session.query(Customer)
|
|
.filter(Customer.active.is_(True))
|
|
.order_by(Customer.name.asc())
|
|
.all()
|
|
)
|
|
customer_items = [{"id": int(c.id), "name": c.name or ""} for c in customers]
|
|
|
|
return render_template(
|
|
"main/reports_new.html",
|
|
initial_customers=customer_items,
|
|
columns_meta=build_report_columns_meta(),
|
|
job_filters_meta=build_report_job_filters_meta(),
|
|
is_edit=False,
|
|
initial_report=None,
|
|
)
|
|
|
|
|
|
@main_bp.route("/reports/<int:report_id>/edit")
|
|
@login_required
|
|
def reports_edit(report_id: int):
|
|
# Editing reports is limited to the same roles that can create them.
|
|
if get_active_role() not in ("admin", "operator", "reporter"):
|
|
return abort(403)
|
|
|
|
r = ReportDefinition.query.get_or_404(report_id)
|
|
|
|
customers = (
|
|
db.session.query(Customer)
|
|
.filter(Customer.active.is_(True))
|
|
.order_by(Customer.name.asc())
|
|
.all()
|
|
)
|
|
customer_items = [{"id": int(c.id), "name": c.name or ""} for c in customers]
|
|
|
|
return render_template(
|
|
"main/reports_new.html",
|
|
initial_customers=customer_items,
|
|
columns_meta=build_report_columns_meta(),
|
|
job_filters_meta=build_report_job_filters_meta(),
|
|
is_edit=True,
|
|
initial_report=_build_report_item(r),
|
|
)
|