Add M365 mapping/prices export-import, license pricing, and PDF summary export
- Mapping (Username/QWE Client/IsRestaurant): CSV export and import via new routes - License prices: persistent DB table (M365LicensePrice), editable price card with auto-save, CSV export/import - Summary table: Monthly Cost column per client + grand-total row, calculated live in JS from current prices - PDF export: client-side jsPDF/autotable, landscape A4, includes cost column when prices are set Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
+205
-4
@@ -8,12 +8,12 @@ import secrets
|
||||
from collections import defaultdict
|
||||
|
||||
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
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 M365UserOverride, get_db
|
||||
from app.database import M365LicensePrice, M365UserOverride, get_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -25,6 +25,14 @@ 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:
|
||||
@@ -32,6 +40,11 @@ def _csrf_token(request: Request) -> str:
|
||||
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):
|
||||
@@ -158,6 +171,10 @@ def _render(request: Request, user, csrf: str, **kwargs):
|
||||
"users_json": "[]",
|
||||
"filename": None,
|
||||
"error": None,
|
||||
"import_success": None,
|
||||
"prices": {},
|
||||
"license_keys": LICENSE_KEYS,
|
||||
"prices_import_success": None,
|
||||
**kwargs,
|
||||
})
|
||||
|
||||
@@ -175,12 +192,28 @@ async def m365_index(request: Request):
|
||||
|
||||
|
||||
@router.get("/m365/license-calculator", response_class=HTMLResponse)
|
||||
async def license_calculator_page(request: Request):
|
||||
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)
|
||||
return _render(request, user, csrf)
|
||||
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)
|
||||
@@ -209,6 +242,7 @@ async def upload_csv(
|
||||
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)
|
||||
@@ -242,3 +276,170 @@ async def save_override(request: Request, db: Session = Depends(get_db)):
|
||||
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user