import re import secrets from datetime import datetime, timezone from fastapi import HTTPException, status from pydantic import BaseModel, Field from pwdlib import PasswordHash from sqlalchemy import select from sqlalchemy.orm import Session from .models import UrmiportReport, UrmiportReportGroup, UserAccount password_hasher = PasswordHash.recommended() GROUP_SLUG_RE = re.compile(r"^[a-z][a-z0-9-]{1,48}[a-z0-9]$") REPORT_ID_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]{0,119}$") class UpsertUrmiportReportPayload(BaseModel): report_group: str = Field(min_length=3, max_length=50) report_id: str | None = Field(default=None, max_length=120) body: str = Field(min_length=1) access: str = Field(default="public", pattern="^(public|password)$") password: str | None = Field(default=None, max_length=120) class CreateUrmiportGroupPayload(BaseModel): report_group: str = Field(min_length=3, max_length=50) display_name: str | None = Field(default=None, max_length=120) default_access: str = Field(default="public", pattern="^(public|password)$") password: str | None = Field(default=None, max_length=120) id_mode: str = Field(default="auto", pattern="^(auto|manual)$") class UpdateUrmiportGroupPayload(BaseModel): display_name: str | None = Field(default=None, max_length=120) default_access: str | None = Field(default=None, pattern="^(public|password)$") password: str | None = Field(default=None, max_length=120) id_mode: str | None = Field(default=None, pattern="^(auto|manual)$") def normalize_group_slug(value: str) -> str: slug = value.strip().lower() if not GROUP_SLUG_RE.match(slug) or "--" in slug: raise ValueError("Report group must be 3-50 chars: lowercase letters, numbers, hyphens") return slug def normalize_report_id(value: str) -> str: report_id = value.strip() if not REPORT_ID_RE.match(report_id): raise ValueError("Report id may contain letters, numbers, dots, underscores, and hyphens") return report_id def hash_password(value: str | None) -> str | None: if not value: return None return password_hasher.hash(value) def verify_password(value: str, hashed: str | None) -> bool: if not hashed: return False return password_hasher.verify(value, hashed) def report_public_url(group_slug: str, report_id: str) -> str: return f"https://urmiai.com/reports/{group_slug}/{report_id}" def serialize_group(group: UrmiportReportGroup, report_count: int = 0) -> dict: return { "id": str(group.id), "report_group": group.slug, "display_name": group.display_name, "default_access": group.default_access, "password_protected": group.default_access == "password", "id_mode": group.id_mode, "report_count": report_count, "url_pattern": f"https://urmiai.com/reports/{group.slug}/{{report_id}}", "created_at": group.created_at, } def serialize_report(report: UrmiportReport, group: UrmiportReportGroup) -> dict: return { "id": str(report.id), "report_group": group.slug, "report_id": report.report_id, "access": report.access_mode, "password_protected": report.access_mode == "password", "url": report_public_url(group.slug, report.report_id), "updated_at": report.updated_at, "created_at": report.created_at, } def get_user_group(db: Session, user: UserAccount, slug: str) -> UrmiportReportGroup: normalized = normalize_group_slug(slug) group = db.scalar( select(UrmiportReportGroup).where( UrmiportReportGroup.user_id == user.id, UrmiportReportGroup.slug == normalized, ) ) if not group: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Report group not found") return group def get_group_by_slug(db: Session, slug: str) -> UrmiportReportGroup: try: normalized = normalize_group_slug(slug) except ValueError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Report group not found") from exc group = db.scalar(select(UrmiportReportGroup).where(UrmiportReportGroup.slug == normalized)) if not group: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Report group not found") return group def get_report(db: Session, group: UrmiportReportGroup, report_id: str) -> UrmiportReport: try: normalized = normalize_report_id(report_id) except ValueError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Report not found") from exc report = db.scalar( select(UrmiportReport).where( UrmiportReport.group_id == group.id, UrmiportReport.report_id == normalized, ) ) if not report: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Report not found") return report def generate_report_id(db: Session, group: UrmiportReportGroup) -> str: for _ in range(12): candidate = f"run-{secrets.token_hex(3)}" exists = db.scalar( select(UrmiportReport.id).where( UrmiportReport.group_id == group.id, UrmiportReport.report_id == candidate, ) ) if not exists: return candidate raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Unable to generate report id") def upsert_report_for_user( db: Session, user: UserAccount, payload: UpsertUrmiportReportPayload, ) -> dict: try: group_slug = normalize_group_slug(payload.report_group) except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc group = db.scalar( select(UrmiportReportGroup).where( UrmiportReportGroup.user_id == user.id, UrmiportReportGroup.slug == group_slug, ) ) if not group: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Report group not found. Create it in Console → Agent → Urmiport first.", ) if payload.report_id: try: report_id = normalize_report_id(payload.report_id) except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc elif group.id_mode == "manual": raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="report_id is required for this report group", ) else: report_id = generate_report_id(db, group) access_mode = payload.access password_hash_value = ( hash_password(payload.password) if access_mode == "password" else None ) if access_mode == "password" and not password_hash_value and not group.password_hash: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="password is required when access is password", ) if access_mode == "password" and not password_hash_value: password_hash_value = group.password_hash existing = db.scalar( select(UrmiportReport).where( UrmiportReport.group_id == group.id, UrmiportReport.report_id == report_id, ) ) if existing: existing.html_body = payload.body existing.access_mode = access_mode existing.password_hash = password_hash_value existing.updated_at = datetime.now(timezone.utc) report = existing else: report = UrmiportReport( user_id=user.id, group_id=group.id, report_id=report_id, html_body=payload.body, access_mode=access_mode, password_hash=password_hash_value, ) db.add(report) db.commit() db.refresh(report) result = serialize_report(report, group) result["success"] = True return result def build_urmiport_api_docs() -> dict: return { "id": "urmiport-upsert-report", "method": "POST", "path": "/api/v1/urmiport/reports", "title": "Publish Urmiport HTML report", "summary": ( "Create or update a hosted HTML report. Reports are served at " "https://urmiai.com/reports/{report_group}/{report_id} without Urmi branding." ), "authentication": { "type": "api_key", "header": "Authorization", "format": "Bearer urmi_live_", "alternate_header": "X-Api-Key", }, "request_body": { "report_group": { "type": "string", "required": True, "description": "Report group slug created in Console → Agent → Urmiport", "example": "acme-agent-runs", }, "report_id": { "type": "string", "required": False, "description": "Report postfix. Required when group id_mode is manual; auto-generated when omitted for auto groups", "example": "run-abc123", }, "body": { "type": "string", "required": True, "description": "Full HTML document or HTML fragment to render on the public report URL", "example": "

Agent run complete

All checks passed.

", }, "access": { "type": "string", "required": False, "description": "public or password", "example": "public", }, "password": { "type": "string", "required": False, "description": "Required when access is password (unless group default password is set)", "example": "secret-pass", }, }, "responses": { "200": { "description": "Report created or updated", "example": { "success": True, "report_group": "acme-agent-runs", "report_id": "run-abc123", "url": "https://urmiai.com/reports/acme-agent-runs/run-abc123", "access": "public", }, }, "401": {"description": "Missing or invalid API key"}, "404": {"description": "Report group not found"}, "422": {"description": "Validation error"}, }, "notes": [ "Create the report group in the Urmi console before calling this API.", "Upsert semantics: same report_group + report_id updates the existing HTML body.", "The hosted page renders only your HTML — no Urmi chrome or branding.", ], }