cgasserandClaude Sonnet 4.6
adf303ad53
New top-nav section with a CSV-based license calculator. Upload an Azure AD user export to see license counts (M365 Standard/Basic, Defender P1, PBI Pro, PBI Premium) grouped by QWE Client. QWE Client and IsRestaurant are editable per user, saved to SQLite, and restored on the next upload. Department-based defaults pre-populate QWE Client (AT→QWE AT, SK→QWE SK) when no saved value exists. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
245 lines
8.2 KiB
Python
245 lines
8.2 KiB
Python
import csv
|
|
import datetime
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import secrets
|
|
from collections import defaultdict
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import get_current_user
|
|
from app.database import M365UserOverride, get_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BASE_DIR = os.path.dirname(__file__)
|
|
templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
|
|
|
|
router = APIRouter()
|
|
|
|
AT_DEPTS = {"QWE AT", "KFC AT", "QWE"}
|
|
SK_DEPTS = {"QWE SK", "KFC SK"}
|
|
|
|
|
|
def _csrf_token(request: Request) -> str:
|
|
if "csrf_token" not in request.session:
|
|
request.session["csrf_token"] = secrets.token_hex(32)
|
|
return request.session["csrf_token"]
|
|
|
|
|
|
def _verify_csrf(request: Request, token: str):
|
|
expected = request.session.get("csrf_token")
|
|
if not expected or not secrets.compare_digest(expected, token):
|
|
raise HTTPException(status_code=403, detail="CSRF token invalid")
|
|
|
|
|
|
def _parse_csv(content: bytes) -> list[dict]:
|
|
try:
|
|
text = content.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
text = content.decode("latin-1")
|
|
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
users = []
|
|
for row in reader:
|
|
upn = row.get("User principal name", "").strip()
|
|
licenses = row.get("Licenses", "").strip()
|
|
department = row.get("Department", "").strip()
|
|
is_licensed = bool(licenses) and licenses != "Unlicensed"
|
|
|
|
company = ""
|
|
if department in AT_DEPTS:
|
|
company = "AT"
|
|
elif department in SK_DEPTS:
|
|
company = "SK"
|
|
|
|
users.append({
|
|
"display_name": row.get("Display name", "").strip(),
|
|
"username": upn,
|
|
"first_name": row.get("First name", "").strip(),
|
|
"last_name": row.get("Last name", "").strip(),
|
|
"department": department,
|
|
"job_title": row.get("Title", "").strip(),
|
|
"usage_location": row.get("Usage location", "").strip(),
|
|
"licenses": licenses,
|
|
"is_licensed": is_licensed,
|
|
"company": company,
|
|
"has_m365_standard": "Microsoft 365 Business Standard" in licenses,
|
|
"has_m365_basic": "Microsoft 365 Business Basic" in licenses,
|
|
"has_defender_p1": "Microsoft Defender for Office 365 (Plan 1)" in licenses,
|
|
"has_pbi_pro": "Power BI Pro" in licenses,
|
|
"has_pbi_premium": "Power BI Premium Per User" in licenses,
|
|
"qwe_client": "QWE AT" if "AT" in department else ("QWE SK" if "SK" in department else ""),
|
|
"is_restaurant": False,
|
|
})
|
|
return users
|
|
|
|
|
|
def _apply_overrides(users: list[dict], db: Session) -> list[dict]:
|
|
upns = [u["username"] for u in users if u["username"]]
|
|
overrides = {
|
|
o.username: o
|
|
for o in db.query(M365UserOverride).filter(M365UserOverride.username.in_(upns)).all()
|
|
}
|
|
for user in users:
|
|
ov = overrides.get(user["username"])
|
|
if ov:
|
|
if ov.qwe_client: # keep department default when saved value is empty
|
|
user["qwe_client"] = ov.qwe_client
|
|
user["is_restaurant"] = bool(ov.is_restaurant)
|
|
return users
|
|
|
|
|
|
def _calculate_summary(users: list[dict]) -> dict:
|
|
licensed = [u for u in users if u["is_licensed"]]
|
|
missing = [u for u in licensed if not u["company"]]
|
|
|
|
# Per-QWE-Client breakdown (primary goal)
|
|
client_groups: dict[str, list] = defaultdict(list)
|
|
for u in licensed:
|
|
key = u["qwe_client"].strip()
|
|
client_groups[key].append(u)
|
|
|
|
def _counts(group: list) -> dict:
|
|
return {
|
|
"m365_standard": sum(1 for u in group if u["has_m365_standard"]),
|
|
"m365_basic": sum(1 for u in group if u["has_m365_basic"]),
|
|
"defender_p1": sum(1 for u in group if u["has_defender_p1"]),
|
|
"pbi_pro": sum(1 for u in group if u["has_pbi_pro"]),
|
|
"pbi_premium": sum(1 for u in group if u["has_pbi_premium"]),
|
|
}
|
|
|
|
# Named clients first (sorted), unnamed last
|
|
client_breakdown = []
|
|
for name in sorted(client_groups.keys(), key=lambda x: (x == "", x.lower())):
|
|
group = client_groups[name]
|
|
row = {"name": name or "(no client)", "count": len(group)}
|
|
row.update(_counts(group))
|
|
client_breakdown.append(row)
|
|
|
|
return {
|
|
"total": len(users),
|
|
"licensed": len(licensed),
|
|
"client_breakdown": client_breakdown,
|
|
"missing": missing,
|
|
}
|
|
|
|
|
|
def _users_for_js(users: list[dict]) -> str:
|
|
"""Minimal user data serialised as JSON for client-side summary recalculation."""
|
|
return json.dumps([
|
|
{
|
|
"username": u["username"],
|
|
"qwe_client": u["qwe_client"],
|
|
"is_licensed": u["is_licensed"],
|
|
"company": u["company"],
|
|
"has_m365_standard": u["has_m365_standard"],
|
|
"has_m365_basic": u["has_m365_basic"],
|
|
"has_defender_p1": u["has_defender_p1"],
|
|
"has_pbi_pro": u["has_pbi_pro"],
|
|
"has_pbi_premium": u["has_pbi_premium"],
|
|
}
|
|
for u in users
|
|
])
|
|
|
|
|
|
def _render(request: Request, user, csrf: str, **kwargs):
|
|
return templates.TemplateResponse("m365_license.html", {
|
|
"request": request,
|
|
"user": user,
|
|
"csrf_token": csrf,
|
|
"users": None,
|
|
"summary": None,
|
|
"users_json": "[]",
|
|
"filename": None,
|
|
"error": None,
|
|
**kwargs,
|
|
})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/m365", response_class=HTMLResponse)
|
|
async def m365_index(request: Request):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse("/auth/login", status_code=302)
|
|
return RedirectResponse("/m365/license-calculator", status_code=302)
|
|
|
|
|
|
@router.get("/m365/license-calculator", response_class=HTMLResponse)
|
|
async def license_calculator_page(request: Request):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse("/auth/login", status_code=302)
|
|
csrf = _csrf_token(request)
|
|
return _render(request, user, csrf)
|
|
|
|
|
|
@router.post("/m365/license-calculator/upload", response_class=HTMLResponse)
|
|
async def upload_csv(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
csrf_token: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse("/auth/login", status_code=302)
|
|
|
|
_verify_csrf(request, csrf_token)
|
|
csrf = _csrf_token(request)
|
|
|
|
try:
|
|
content = await file.read()
|
|
users = _parse_csv(content)
|
|
users = _apply_overrides(users, db)
|
|
summary = _calculate_summary(users)
|
|
licensed = [u for u in users if u["is_licensed"]]
|
|
return _render(
|
|
request, user, csrf,
|
|
users=licensed,
|
|
summary=summary,
|
|
users_json=_users_for_js(licensed),
|
|
filename=file.filename,
|
|
)
|
|
except Exception as exc:
|
|
logger.error("M365 CSV parse error: %s", exc)
|
|
return _render(request, user, csrf, error=f"Failed to process CSV: {exc}")
|
|
|
|
|
|
@router.post("/m365/override")
|
|
async def save_override(request: Request, db: Session = Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
|
|
body = await request.json()
|
|
_verify_csrf(request, body.get("csrf_token", ""))
|
|
|
|
username = body.get("username", "").strip()
|
|
if not username:
|
|
raise HTTPException(status_code=400, detail="username required")
|
|
|
|
override = db.query(M365UserOverride).filter(M365UserOverride.username == username).first()
|
|
if override:
|
|
override.qwe_client = body.get("qwe_client") or ""
|
|
override.is_restaurant = bool(body.get("is_restaurant", False))
|
|
override.updated_at = datetime.datetime.utcnow()
|
|
else:
|
|
db.add(M365UserOverride(
|
|
username=username,
|
|
qwe_client=body.get("qwe_client") or "",
|
|
is_restaurant=bool(body.get("is_restaurant", False)),
|
|
))
|
|
|
|
db.commit()
|
|
return {"status": "ok"}
|