366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Compliance Center — licenses & certificates with expiry tracking."""
|
|
import base64
|
|
import binascii
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, status
|
|
|
|
from app.core.deps import require_org, require_client_roles, Principal
|
|
from app.models import Compliance, Product,AuditLog
|
|
from app.models.documents import now
|
|
from app.models.enums import ClientRole, ComplianceStatus
|
|
from app.services.storage import upload_file
|
|
|
|
router = APIRouter(prefix="/compliance", tags=["compliance"])
|
|
|
|
|
|
def _parse(d):
|
|
if not d:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(str(d).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _status_for(expiry: Optional[datetime], has_number: bool) -> ComplianceStatus:
|
|
today = now()
|
|
|
|
if expiry:
|
|
exp = expiry if expiry.tzinfo else expiry.replace(tzinfo=timezone.utc)
|
|
|
|
if exp < today:
|
|
return ComplianceStatus.expired
|
|
|
|
if exp <= today + timedelta(days=30):
|
|
return ComplianceStatus.expiring
|
|
|
|
if not has_number:
|
|
return ComplianceStatus.pending
|
|
|
|
return ComplianceStatus.complete
|
|
|
|
|
|
def _days_left(expiry: Optional[datetime]):
|
|
if not expiry:
|
|
return None
|
|
exp = expiry if expiry.tzinfo else expiry.replace(tzinfo=timezone.utc)
|
|
return (exp - now()).days
|
|
|
|
|
|
def _out(c: Compliance, prod=None) -> dict:
|
|
return {
|
|
"id": str(c.id), "product_id": c.product_id,
|
|
"product_name": prod.name if prod else None,
|
|
"sku": prod.sku if prod else None,
|
|
"product_image": prod.image_url if prod else None,
|
|
"compliance_type": c.compliance_type,
|
|
"license_number": c.license_number, "license_name": c.license_name,
|
|
"issuing_authority": c.issuing_authority,
|
|
"issue_date": c.issue_date, "expiry_date": c.expiry_date,
|
|
"days_left": _days_left(c.expiry_date),
|
|
"country": c.country, "state": c.state, "plant": c.plant,
|
|
"category": c.category, "applicable_to": c.applicable_to,
|
|
"remarks": c.remarks, "document_url": c.document_url,
|
|
"documents": c.documents, "created_by": c.created_by,
|
|
"status": c.status, "created_at": c.created_at, "updated_at": c.updated_at,
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_compliance(p: Principal = Depends(require_org),
|
|
status_filter: Optional[str] = None,
|
|
compliance_type: Optional[str] = None,
|
|
country: Optional[str] = None,
|
|
category: Optional[str] = None,
|
|
search: Optional[str] = None):
|
|
items = await Compliance.find(Compliance.org_id == str(p.org.id)).to_list()
|
|
products = {str(x.id): x for x in await Product.find(Product.org_id == str(p.org.id)).to_list()}
|
|
for c in items:
|
|
c.status = _status_for(c.expiry_date, bool(c.license_number))
|
|
rows = []
|
|
for c in items:
|
|
if status_filter and c.status != status_filter:
|
|
continue
|
|
if compliance_type and c.compliance_type != compliance_type:
|
|
continue
|
|
if country and c.country != country:
|
|
continue
|
|
if category and c.category != category:
|
|
continue
|
|
prod = products.get(c.product_id or "")
|
|
if search:
|
|
s = search.lower()
|
|
hay = f"{c.compliance_type} {c.license_number or ''} {c.issuing_authority or ''} {prod.name if prod else ''}".lower()
|
|
if s not in hay:
|
|
continue
|
|
rows.append(_out(c, prod))
|
|
rows.sort(key=lambda r: r["updated_at"], reverse=True)
|
|
return rows
|
|
|
|
|
|
@router.get("/summary")
|
|
async def summary(p: Principal = Depends(require_org)):
|
|
items = await Compliance.find(Compliance.org_id == str(p.org.id)).to_list()
|
|
for c in items:
|
|
c.status = _status_for(c.expiry_date, bool(c.license_number))
|
|
def n(s): return sum(1 for c in items if c.status == s)
|
|
# Match the dashboard's Products KPI: use the org's live product_count
|
|
# (excludes archived), not a raw collection count.
|
|
total_products = p.org.product_count
|
|
total = len(items) or 1
|
|
return {
|
|
"total": len(items),
|
|
"total_products": total_products,
|
|
"complete": n(ComplianceStatus.complete),
|
|
"pending": n(ComplianceStatus.pending),
|
|
"expired": n(ComplianceStatus.expired),
|
|
"expiring": n(ComplianceStatus.expiring),
|
|
"complete_pct": round(n(ComplianceStatus.complete) / total * 100, 1),
|
|
"pending_pct": round(n(ComplianceStatus.pending) / total * 100, 1),
|
|
"expired_pct": round(n(ComplianceStatus.expired) / total * 100, 1),
|
|
"alerts": [
|
|
{"type": c.compliance_type, "number": c.license_number,
|
|
"status": c.status, "expiry": c.expiry_date}
|
|
for c in items if c.status in (ComplianceStatus.expired, ComplianceStatus.expiring)
|
|
],
|
|
}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_compliance(body: dict = Body(...),
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
expiry = _parse(body.get("expiry_date"))
|
|
c = Compliance(
|
|
org_id=str(p.org.id), product_id=body.get("product_id"),
|
|
compliance_type=body.get("compliance_type", "Certificate"),
|
|
license_number=body.get("license_number"),
|
|
license_name=body.get("license_name"),
|
|
issuing_authority=body.get("issuing_authority"),
|
|
issue_date=_parse(body.get("issue_date")), expiry_date=expiry,
|
|
country=body.get("country"), state=body.get("state"), plant=body.get("plant"),
|
|
category=body.get("category"), applicable_to=body.get("applicable_to"),
|
|
remarks=body.get("remarks"), created_by=p.user.name,
|
|
status=_status_for(expiry, bool(body.get("license_number"))),
|
|
)
|
|
await c.insert()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="compliance.create",
|
|
module="compliance",
|
|
details=f"Created {c.compliance_type} for product {c.product_id}",
|
|
).insert()
|
|
return _out(c, await _safe_prod(c.product_id))
|
|
|
|
|
|
async def _safe_prod(pid):
|
|
if not pid:
|
|
return None
|
|
try:
|
|
return await Product.get(pid)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def _owned(cid: str, p: Principal) -> Compliance:
|
|
try:
|
|
c = await Compliance.get(cid)
|
|
except Exception:
|
|
c = None
|
|
if not c or c.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Compliance record not found")
|
|
return c
|
|
|
|
|
|
@router.get("/{cid}")
|
|
async def get_compliance(cid: str, p: Principal = Depends(require_org)):
|
|
c = await _owned(cid, p)
|
|
return _out(c, await _safe_prod(c.product_id))
|
|
|
|
|
|
@router.get("/{cid}/detail")
|
|
async def compliance_detail(cid: str, p: Principal = Depends(require_org)):
|
|
"""Full detail for the Compliance Details modal (overview + documents + history)."""
|
|
c = await _owned(cid, p)
|
|
return {
|
|
**_out(c, await _safe_prod(c.product_id)),
|
|
"history": [
|
|
{"action": "Created", "by": c.created_by, "at": c.created_at},
|
|
{"action": "Last updated", "by": c.created_by, "at": c.updated_at},
|
|
],
|
|
}
|
|
|
|
|
|
@router.put("/{cid}")
|
|
async def update_compliance(cid: str, body: dict = Body(...),
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
c = await _owned(cid, p)
|
|
for k in ("compliance_type", "license_number", "license_name", "issuing_authority",
|
|
"country", "state", "plant", "category", "applicable_to", "remarks"):
|
|
if k in body:
|
|
setattr(c, k, body[k])
|
|
if "issue_date" in body:
|
|
c.issue_date = _parse(body["issue_date"])
|
|
if "expiry_date" in body:
|
|
c.expiry_date = _parse(body["expiry_date"])
|
|
c.status = _status_for(c.expiry_date, bool(c.license_number))
|
|
c.updated_at = now()
|
|
await c.save()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="compliance.update",
|
|
module="compliance",
|
|
details=f"Updated {c.compliance_type}",
|
|
).insert()
|
|
return _out(c, await _safe_prod(c.product_id))
|
|
|
|
|
|
|
|
@router.delete("/{cid}", status_code=204)
|
|
async def delete_compliance(cid: str,
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin))):
|
|
c = await _owned(cid, p)
|
|
await c.delete()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="compliance.delete",
|
|
module="compliance",
|
|
details=f"Deleted {c.compliance_type}",
|
|
).insert()
|
|
|
|
|
|
@router.post("/{cid}/document")
|
|
async def upload_document(cid: str, body: dict = Body(...),
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
"""body = {data_url, filename}"""
|
|
c = await _owned(cid, p)
|
|
data_url = body.get("data_url", "")
|
|
if "," not in data_url:
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid data URL")
|
|
try:
|
|
content = base64.b64decode(data_url.split(",", 1)[1])
|
|
except (binascii.Error, ValueError):
|
|
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Bad base64")
|
|
fname = body.get("filename", f"{cid}.pdf")
|
|
url = await upload_file(content, fname, "application/pdf")
|
|
c.document_url = url
|
|
c.documents = (c.documents or []) + [{"name": fname, "url": url,
|
|
"uploaded_at": now().isoformat()}]
|
|
c.updated_at = now()
|
|
await c.save()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="compliance.upload_document",
|
|
module="compliance",
|
|
details=f"Uploaded document {fname}",
|
|
).insert()
|
|
return {"document_url": url, "documents": c.documents}
|
|
|
|
|
|
|
|
import csv
|
|
import io
|
|
|
|
from fastapi import UploadFile, File
|
|
|
|
@router.post("/import")
|
|
async def import_compliance(
|
|
file: UploadFile = File(...),
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user)),
|
|
):
|
|
if not file.filename.lower().endswith(".csv"):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Only CSV files are supported."
|
|
)
|
|
|
|
content = await file.read()
|
|
|
|
try:
|
|
text = content.decode("utf-8-sig")
|
|
except Exception:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid CSV file."
|
|
)
|
|
|
|
reader = csv.DictReader(io.StringIO(text))
|
|
|
|
inserted = 0
|
|
skipped = 0
|
|
|
|
products = await Product.find(
|
|
Product.org_id == str(p.org.id)
|
|
).to_list()
|
|
|
|
product_map = {
|
|
x.name.strip().lower(): str(x.id)
|
|
for x in products
|
|
}
|
|
|
|
for row in reader:
|
|
product_name = (row.get("Product") or "").strip()
|
|
|
|
product_id = product_map.get(product_name.lower())
|
|
|
|
if not product_id:
|
|
skipped += 1
|
|
continue
|
|
|
|
expiry = _parse(row.get("Expiry Date"))
|
|
issue = _parse(row.get("Issue Date"))
|
|
|
|
compliance = Compliance(
|
|
org_id=str(p.org.id),
|
|
product_id=product_id,
|
|
compliance_type=row.get("Compliance Type"),
|
|
license_number=row.get("License Number"),
|
|
license_name=row.get("License Name"),
|
|
issuing_authority=row.get("Issuing Authority"),
|
|
issue_date=issue,
|
|
expiry_date=expiry,
|
|
country=row.get("Country"),
|
|
state=row.get("State"),
|
|
category=row.get("Category"),
|
|
applicable_to=row.get("Applicable To"),
|
|
plant=row.get("Manufacturing Plant"),
|
|
remarks=row.get("Remarks"),
|
|
created_by=p.user.name,
|
|
status=_status_for(
|
|
expiry,
|
|
bool(row.get("License Number"))
|
|
),
|
|
)
|
|
|
|
await compliance.insert()
|
|
inserted += 1
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="compliance.import",
|
|
module="compliance",
|
|
details=f"Imported {inserted} compliance records",
|
|
).insert()
|
|
|
|
return {
|
|
"message": "Import completed.",
|
|
"inserted": inserted,
|
|
"skipped": skipped,
|
|
}
|
|
|