cgasserandClaude Sonnet 4.6
8af2fe75a6
- Total row now sums each license column individually instead of using colspan - Missing Assignment warning now flags licensed users with no QWE Client set, replacing the previous department-based company check Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
446 lines
15 KiB
Python
446 lines
15 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, StreamingResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import get_current_user
|
|
from app.database import M365LicensePrice, 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"}
|
|
|
|
LICENSE_KEYS = [
|
|
("m365_standard", "M365 Business Standard"),
|
|
("m365_basic", "M365 Business Basic"),
|
|
("defender_p1", "Defender P1"),
|
|
("pbi_pro", "Power BI Pro"),
|
|
("pbi_premium", "Power BI Premium"),
|
|
]
|
|
|
|
|
|
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 _get_prices(db: Session) -> dict[str, float]:
|
|
rows = {r.license_key: r.price for r in db.query(M365LicensePrice).all()}
|
|
return {key: rows.get(key, 0.0) for key, _ in LICENSE_KEYS}
|
|
|
|
|
|
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["qwe_client"].strip()]
|
|
|
|
# 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,
|
|
"import_success": None,
|
|
"prices": {},
|
|
"license_keys": LICENSE_KEYS,
|
|
"prices_import_success": 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,
|
|
imported: int | None = None,
|
|
import_error: bool = False,
|
|
prices_imported: int | None = None,
|
|
prices_error: bool = False,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse("/auth/login", status_code=302)
|
|
csrf = _csrf_token(request)
|
|
kwargs: dict = {"prices": _get_prices(db)}
|
|
if imported is not None:
|
|
kwargs["import_success"] = imported
|
|
if import_error:
|
|
kwargs["error"] = "Failed to import mapping CSV. Check that the file has Username, QWE Client, and IsRestaurant columns."
|
|
if prices_imported is not None:
|
|
kwargs["prices_import_success"] = prices_imported
|
|
if prices_error:
|
|
kwargs["error"] = "Failed to import prices CSV. Check that the file has License Key and Price columns."
|
|
return _render(request, user, csrf, **kwargs)
|
|
|
|
|
|
@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,
|
|
prices=_get_prices(db),
|
|
)
|
|
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"}
|
|
|
|
|
|
@router.get("/m365/mapping/export")
|
|
async def export_mapping(request: Request, db: Session = Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse("/auth/login", status_code=302)
|
|
|
|
overrides = db.query(M365UserOverride).order_by(M365UserOverride.username).all()
|
|
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow(["Username", "QWE Client", "IsRestaurant"])
|
|
for o in overrides:
|
|
writer.writerow([o.username, o.qwe_client, "1" if o.is_restaurant else "0"])
|
|
|
|
return StreamingResponse(
|
|
iter([output.getvalue()]),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=m365_mapping.csv"},
|
|
)
|
|
|
|
|
|
@router.post("/m365/mapping/import", response_class=HTMLResponse)
|
|
async def import_mapping(
|
|
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)
|
|
|
|
try:
|
|
content = await file.read()
|
|
try:
|
|
text = content.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
text = content.decode("latin-1")
|
|
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
count = 0
|
|
for row in reader:
|
|
username = row.get("Username", "").strip()
|
|
if not username:
|
|
continue
|
|
qwe_client = row.get("QWE Client", "").strip()
|
|
is_restaurant = row.get("IsRestaurant", "0").strip().lower() in ("1", "true", "yes")
|
|
|
|
override = db.query(M365UserOverride).filter(M365UserOverride.username == username).first()
|
|
if override:
|
|
override.qwe_client = qwe_client
|
|
override.is_restaurant = is_restaurant
|
|
override.updated_at = datetime.datetime.utcnow()
|
|
else:
|
|
db.add(M365UserOverride(
|
|
username=username,
|
|
qwe_client=qwe_client,
|
|
is_restaurant=is_restaurant,
|
|
))
|
|
count += 1
|
|
|
|
db.commit()
|
|
return RedirectResponse(f"/m365/license-calculator?imported={count}", status_code=302)
|
|
except Exception as exc:
|
|
logger.error("M365 mapping import error: %s", exc)
|
|
return RedirectResponse("/m365/license-calculator?import_error=1", status_code=302)
|
|
|
|
|
|
@router.post("/m365/prices")
|
|
async def save_prices(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", ""))
|
|
|
|
for key, _ in LICENSE_KEYS:
|
|
raw = body.get(key)
|
|
if raw is None:
|
|
continue
|
|
try:
|
|
price = max(0.0, float(raw))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
|
|
row = db.query(M365LicensePrice).filter(M365LicensePrice.license_key == key).first()
|
|
if row:
|
|
row.price = price
|
|
row.updated_at = datetime.datetime.utcnow()
|
|
else:
|
|
db.add(M365LicensePrice(license_key=key, price=price))
|
|
|
|
db.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/m365/prices/export")
|
|
async def export_prices(request: Request, db: Session = Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse("/auth/login", status_code=302)
|
|
|
|
prices = _get_prices(db)
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow(["License Key", "Label", "Price"])
|
|
for key, label in LICENSE_KEYS:
|
|
writer.writerow([key, label, prices[key]])
|
|
|
|
return StreamingResponse(
|
|
iter([output.getvalue()]),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=m365_prices.csv"},
|
|
)
|
|
|
|
|
|
@router.post("/m365/prices/import", response_class=HTMLResponse)
|
|
async def import_prices(
|
|
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)
|
|
|
|
valid_keys = {key for key, _ in LICENSE_KEYS}
|
|
|
|
try:
|
|
content = await file.read()
|
|
try:
|
|
text = content.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
text = content.decode("latin-1")
|
|
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
count = 0
|
|
for row in reader:
|
|
key = row.get("License Key", "").strip()
|
|
if key not in valid_keys:
|
|
continue
|
|
try:
|
|
price = max(0.0, float(row.get("Price", "0").strip()))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
|
|
existing = db.query(M365LicensePrice).filter(M365LicensePrice.license_key == key).first()
|
|
if existing:
|
|
existing.price = price
|
|
existing.updated_at = datetime.datetime.utcnow()
|
|
else:
|
|
db.add(M365LicensePrice(license_key=key, price=price))
|
|
count += 1
|
|
|
|
db.commit()
|
|
return RedirectResponse(f"/m365/license-calculator?prices_imported={count}", status_code=302)
|
|
except Exception as exc:
|
|
logger.error("M365 prices import error: %s", exc)
|
|
return RedirectResponse("/m365/license-calculator?prices_error=1", status_code=302)
|