f909182632
All checks were successful
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-ai-compliance (push) Successful in 37s
CI / test-python-backend-compliance (push) Successful in 32s
CI / test-python-document-crawler (push) Successful in 21s
CI / test-python-dsms-gateway (push) Successful in 18s
Implementiert MIT-lizenzierte DSGVO-Templates (DSE, Impressum, AGB) in der eigenen PostgreSQL-Datenbank statt KLAUSUR_SERVICE-Abhängigkeit. - Migration 018: compliance_legal_templates Tabelle + 3 Seed-Templates - Routes: GET/POST/PUT/DELETE /legal-templates + /status + /sources - Registriert im bestehenden compliance catch-all Proxy (kein neuer Proxy) - searchTemplates.ts: eigenes Backend als Primary, RAG bleibt Fallback - ServiceMode-Banner: KLAUSUR_SERVICE-Referenz entfernt - Tests: 25 Python + 3 Vitest — alle grün Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""
|
|
FastAPI routes for Legal Templates (Document Generator).
|
|
|
|
Self-authored templates (MIT License) stored in compliance_legal_templates.
|
|
|
|
Endpoints:
|
|
GET /legal-templates — list (query, document_type, language, status, limit, offset)
|
|
GET /legal-templates/status — counts by type
|
|
GET /legal-templates/sources — distinct source_name list
|
|
GET /legal-templates/{id} — single template by id
|
|
POST /legal-templates — create (admin)
|
|
PUT /legal-templates/{id} — update
|
|
DELETE /legal-templates/{id} — delete (204)
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional, List, Any, Dict
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Header
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
from uuid import UUID
|
|
|
|
from classroom_engine.database import get_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/legal-templates", tags=["legal-templates"])
|
|
|
|
DEFAULT_TENANT_ID = "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|
|
|
|
VALID_DOCUMENT_TYPES = {"privacy_policy", "terms_of_service", "impressum"}
|
|
VALID_STATUSES = {"published", "draft", "archived"}
|
|
|
|
|
|
# =============================================================================
|
|
# Pydantic Schemas
|
|
# =============================================================================
|
|
|
|
class LegalTemplateCreate(BaseModel):
|
|
document_type: str
|
|
title: str
|
|
description: Optional[str] = None
|
|
content: str
|
|
placeholders: Optional[List[str]] = None
|
|
language: str = "de"
|
|
jurisdiction: str = "DE"
|
|
license_id: str = "mit"
|
|
license_name: str = "MIT License"
|
|
source_name: str = "BreakPilot Compliance"
|
|
attribution_required: bool = False
|
|
is_complete_document: bool = True
|
|
version: str = "1.0.0"
|
|
status: str = "published"
|
|
|
|
|
|
class LegalTemplateUpdate(BaseModel):
|
|
document_type: Optional[str] = None
|
|
title: Optional[str] = None
|
|
description: Optional[str] = None
|
|
content: Optional[str] = None
|
|
placeholders: Optional[List[str]] = None
|
|
language: Optional[str] = None
|
|
jurisdiction: Optional[str] = None
|
|
license_id: Optional[str] = None
|
|
license_name: Optional[str] = None
|
|
source_name: Optional[str] = None
|
|
attribution_required: Optional[bool] = None
|
|
is_complete_document: Optional[bool] = None
|
|
version: Optional[str] = None
|
|
status: Optional[str] = None
|
|
|
|
|
|
# =============================================================================
|
|
# Helpers
|
|
# =============================================================================
|
|
|
|
def _row_to_dict(row) -> Dict[str, Any]:
|
|
result = dict(row._mapping)
|
|
for key, val in result.items():
|
|
if isinstance(val, datetime):
|
|
result[key] = val.isoformat()
|
|
elif hasattr(val, '__str__') and not isinstance(val, (str, int, float, bool, list, dict, type(None))):
|
|
result[key] = str(val)
|
|
return result
|
|
|
|
|
|
def _get_tenant_id(x_tenant_id: Optional[str] = Header(None)) -> str:
|
|
if x_tenant_id:
|
|
try:
|
|
UUID(x_tenant_id)
|
|
return x_tenant_id
|
|
except ValueError:
|
|
pass
|
|
return DEFAULT_TENANT_ID
|
|
|
|
|
|
# =============================================================================
|
|
# Routes
|
|
# =============================================================================
|
|
|
|
@router.get("")
|
|
async def list_legal_templates(
|
|
query: Optional[str] = Query(None, description="Full-text ILIKE search on title/description/content"),
|
|
document_type: Optional[str] = Query(None),
|
|
language: Optional[str] = Query(None),
|
|
status: Optional[str] = Query("published"),
|
|
limit: int = Query(50, ge=1, le=200),
|
|
offset: int = Query(0, ge=0),
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""List legal templates with optional filters."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
|
|
where_clauses = ["tenant_id = :tenant_id"]
|
|
params: Dict[str, Any] = {"tenant_id": tenant_id, "limit": limit, "offset": offset}
|
|
|
|
if status:
|
|
where_clauses.append("status = :status")
|
|
params["status"] = status
|
|
if document_type:
|
|
where_clauses.append("document_type = :document_type")
|
|
params["document_type"] = document_type
|
|
if language:
|
|
where_clauses.append("language = :language")
|
|
params["language"] = language
|
|
if query:
|
|
where_clauses.append(
|
|
"(title ILIKE :query OR description ILIKE :query OR content ILIKE :query)"
|
|
)
|
|
params["query"] = f"%{query}%"
|
|
|
|
where_sql = " AND ".join(where_clauses)
|
|
|
|
total_row = db.execute(
|
|
text(f"SELECT COUNT(*) FROM compliance_legal_templates WHERE {where_sql}"),
|
|
params,
|
|
).fetchone()
|
|
total = total_row[0] if total_row else 0
|
|
|
|
rows = db.execute(
|
|
text(f"""
|
|
SELECT * FROM compliance_legal_templates
|
|
WHERE {where_sql}
|
|
ORDER BY document_type, title
|
|
LIMIT :limit OFFSET :offset
|
|
"""),
|
|
params,
|
|
).fetchall()
|
|
|
|
return {
|
|
"templates": [_row_to_dict(r) for r in rows],
|
|
"total": total,
|
|
}
|
|
|
|
|
|
@router.get("/status")
|
|
async def get_templates_status(
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""Return template counts by document_type."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
|
|
row = db.execute(text("""
|
|
SELECT
|
|
COUNT(*) AS total,
|
|
COUNT(*) FILTER (WHERE status = 'published') AS published,
|
|
COUNT(*) FILTER (WHERE status = 'draft') AS draft,
|
|
COUNT(*) FILTER (WHERE status = 'archived') AS archived,
|
|
COUNT(*) FILTER (WHERE document_type = 'privacy_policy') AS privacy_policy,
|
|
COUNT(*) FILTER (WHERE document_type = 'terms_of_service') AS terms_of_service,
|
|
COUNT(*) FILTER (WHERE document_type = 'impressum') AS impressum
|
|
FROM compliance_legal_templates
|
|
WHERE tenant_id = :tenant_id
|
|
"""), {"tenant_id": tenant_id}).fetchone()
|
|
|
|
if row:
|
|
d = dict(row._mapping)
|
|
counts = {k: int(v or 0) for k, v in d.items()}
|
|
return {
|
|
"total": counts["total"],
|
|
"by_status": {
|
|
"published": counts["published"],
|
|
"draft": counts["draft"],
|
|
"archived": counts["archived"],
|
|
},
|
|
"by_type": {
|
|
"privacy_policy": counts["privacy_policy"],
|
|
"terms_of_service": counts["terms_of_service"],
|
|
"impressum": counts["impressum"],
|
|
},
|
|
}
|
|
return {"total": 0, "by_status": {}, "by_type": {}}
|
|
|
|
|
|
@router.get("/sources")
|
|
async def get_template_sources(
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""Return distinct source_name values."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
|
|
rows = db.execute(
|
|
text("SELECT DISTINCT source_name FROM compliance_legal_templates WHERE tenant_id = :tenant_id ORDER BY source_name"),
|
|
{"tenant_id": tenant_id},
|
|
).fetchall()
|
|
|
|
return {"sources": [r[0] for r in rows]}
|
|
|
|
|
|
@router.get("/{template_id}")
|
|
async def get_legal_template(
|
|
template_id: str,
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""Fetch a single template by ID."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
row = db.execute(
|
|
text("SELECT * FROM compliance_legal_templates WHERE id = :id AND tenant_id = :tenant_id"),
|
|
{"id": template_id, "tenant_id": tenant_id},
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Template not found")
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_legal_template(
|
|
payload: LegalTemplateCreate,
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""Create a new legal template."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
|
|
if payload.document_type not in VALID_DOCUMENT_TYPES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid document_type. Must be one of: {', '.join(VALID_DOCUMENT_TYPES)}"
|
|
)
|
|
if payload.status not in VALID_STATUSES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid status. Must be one of: {', '.join(VALID_STATUSES)}"
|
|
)
|
|
|
|
placeholders_json = json.dumps(payload.placeholders or [])
|
|
|
|
row = db.execute(text("""
|
|
INSERT INTO compliance_legal_templates (
|
|
tenant_id, document_type, title, description, content,
|
|
placeholders, language, jurisdiction,
|
|
license_id, license_name, source_name,
|
|
attribution_required, is_complete_document, version, status
|
|
) VALUES (
|
|
:tenant_id, :document_type, :title, :description, :content,
|
|
CAST(:placeholders AS jsonb), :language, :jurisdiction,
|
|
:license_id, :license_name, :source_name,
|
|
:attribution_required, :is_complete_document, :version, :status
|
|
) RETURNING *
|
|
"""), {
|
|
"tenant_id": tenant_id,
|
|
"document_type": payload.document_type,
|
|
"title": payload.title,
|
|
"description": payload.description,
|
|
"content": payload.content,
|
|
"placeholders": placeholders_json,
|
|
"language": payload.language,
|
|
"jurisdiction": payload.jurisdiction,
|
|
"license_id": payload.license_id,
|
|
"license_name": payload.license_name,
|
|
"source_name": payload.source_name,
|
|
"attribution_required": payload.attribution_required,
|
|
"is_complete_document": payload.is_complete_document,
|
|
"version": payload.version,
|
|
"status": payload.status,
|
|
}).fetchone()
|
|
db.commit()
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.put("/{template_id}")
|
|
async def update_legal_template(
|
|
template_id: str,
|
|
payload: LegalTemplateUpdate,
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""Update an existing legal template."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
if not updates:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
|
|
if "document_type" in updates and updates["document_type"] not in VALID_DOCUMENT_TYPES:
|
|
raise HTTPException(status_code=400, detail=f"Invalid document_type")
|
|
if "status" in updates and updates["status"] not in VALID_STATUSES:
|
|
raise HTTPException(status_code=400, detail=f"Invalid status")
|
|
|
|
set_clauses = ["updated_at = :updated_at"]
|
|
params: Dict[str, Any] = {
|
|
"id": template_id,
|
|
"tenant_id": tenant_id,
|
|
"updated_at": datetime.utcnow(),
|
|
}
|
|
|
|
for field, value in updates.items():
|
|
if field == "placeholders":
|
|
params[field] = json.dumps(value if value is not None else [])
|
|
set_clauses.append(f"{field} = CAST(:{field} AS jsonb)")
|
|
else:
|
|
params[field] = value
|
|
set_clauses.append(f"{field} = :{field}")
|
|
|
|
row = db.execute(
|
|
text(f"""
|
|
UPDATE compliance_legal_templates
|
|
SET {', '.join(set_clauses)}
|
|
WHERE id = :id AND tenant_id = :tenant_id
|
|
RETURNING *
|
|
"""),
|
|
params,
|
|
).fetchone()
|
|
db.commit()
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Template not found")
|
|
return _row_to_dict(row)
|
|
|
|
|
|
@router.delete("/{template_id}", status_code=204)
|
|
async def delete_legal_template(
|
|
template_id: str,
|
|
db: Session = Depends(get_db),
|
|
x_tenant_id: Optional[str] = Header(None),
|
|
):
|
|
"""Delete a legal template."""
|
|
tenant_id = _get_tenant_id(x_tenant_id)
|
|
result = db.execute(
|
|
text("DELETE FROM compliance_legal_templates WHERE id = :id AND tenant_id = :tenant_id"),
|
|
{"id": template_id, "tenant_id": tenant_id},
|
|
)
|
|
db.commit()
|
|
if result.rowcount == 0:
|
|
raise HTTPException(status_code=404, detail="Template not found")
|