Feature/dialplan editor

app/api/telephoning.py

@@ -5,31 +5,32 @@
 Licensed under the MIT license. See LICENSE file in the project root for details.
 """
 
-from asterisk.ami.response import FutureResponse
-from logging import getLogger
-from asterisk.ami.client import AMIClientAdapter
-from app.models.crud.asterisk import get_contact
-from app.core.db import SessionDep
-from app.models.crud.extension import get_extension_by_id
 from datetime import datetime
+from logging import getLogger
 from typing import Optional
-from fastapi import APIRouter, HTTPException
-from fastapi import status
+
+from asterisk.ami.client import AMIClientAdapter
+from asterisk.ami.response import FutureResponse
+from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
 from app.api.deps import CurrentUser, OptionalCurrentUser
-from app.core.db import SessionAsteriskDep
+from app.core.config import settings
+from app.core.db import SessionAsteriskDep, SessionDep
+from app.models.crud import CRUDNotAllowedException
+from app.models.crud.asterisk import get_contact, get_known_dialplan_extensions
+from app.models.crud.extension import get_extension_by_id
 from app.models.user import UserRole
+from app.telephoning.dialplan import Dialplan
 from app.telephoning.flavor import MediaDescriptor
 from app.telephoning.main import Telephoning
-from app.core.config import settings
 from app.telephoning.websip import WebSIPExtension, WebSIPManager
 
-
 router = APIRouter(prefix="/telephoning", tags=["telephoning"])
 
 logger = getLogger(__name__)
 
+
 class PhoneType(BaseModel):
     schema: Optional[dict] = None
     display_index: int
@@ -107,33 +108,100 @@ def put_websip(extension: str):
 
 
 @router.get("/originate", status_code=status.HTTP_204_NO_CONTENT)
-def originate_call(session: SessionDep, session_asterisk: SessionAsteriskDep, user: CurrentUser, source: str, dest: str):
+def originate_call(
+    session: SessionDep,
+    session_asterisk: SessionAsteriskDep,
+    user: CurrentUser,
+    source: str,
+    dest: str,
+):
     source_extension = get_extension_by_id(session, source, public=False)
     if source_extension is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Source extension unknown")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Source extension unknown"
+        )
     if source_extension.user_id != user.id and not user.role == UserRole.ADMIN:
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You may not originate calls from this extension")
-
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You may not originate calls from this extension",
+        )
+
     contact = get_contact(session_asterisk, source_extension, user)
     if contact is None:
-        raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="The source extension is offline")
+        raise HTTPException(
+            status_code=status.HTTP_406_NOT_ACCEPTABLE,
+            detail="The source extension is offline",
+        )
 
     if not dest.isnumeric():
-        raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE, detail="Destination must be a number!")
+        raise HTTPException(
+            status_code=status.HTTP_406_NOT_ACCEPTABLE,
+            detail="Destination must be a number!",
+        )
 
     client = Telephoning.instance().get_ami_client()
     adapter = AMIClientAdapter(client)
     response: FutureResponse = adapter.Originate(
-        Channel=f"PJSIP/{source}",
-        Exten=dest,
-        Priority=1,
-        Context="pjsip_internal"
+        Channel=f"PJSIP/{source}", Exten=dest, Priority=1, Context="pjsip_internal"
     )
 
     logger.info(f"Originated call from {source} to {dest}")
     logger.debug(f"Received AMI response:\n{response.response}")
 
     if response.response is not None:
-        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="unable to originate call due to error from AMI")
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="unable to originate call due to error from AMI",
+        )
+
+    return {}
 
-    return {}
+
+@router.get("/dialplan/schemas")
+def get_dialplan_application_schemas(user: CurrentUser) -> dict[str, dict]:
+    if user.role != UserRole.ADMIN:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only admins are permitted to request this!",
+        )
+    schemas = {}
+    for app in Dialplan.get_known_apps():
+        schemas.update({app.COMPATIBLE_APP: app.model_json_schema()})
+        schemas[app.COMPATIBLE_APP].update({"doc_url": app.DOC_URL})
+    return schemas
+
+
+@router.post("/dialplan/store")
+def store_dialplan(
+    session_asterisk: SessionAsteriskDep, user: CurrentUser, plan: Dialplan
+):
+    if user.role != UserRole.ADMIN:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only admins are permitted to store dialplans!",
+        )
+    plan.store(session_asterisk)
+    return plan
+
+
+@router.get("/dialplan/{exten}")
+def get_dialplan(
+    session_asterisk: SessionAsteriskDep, user: CurrentUser, exten: str
+) -> Dialplan | None:
+    if user.role != UserRole.ADMIN:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only admins are permitted to request dialplans!",
+        )
+    plan = Dialplan.from_db(session_asterisk, exten)
+    return plan
+
+
+@router.get("/dialplans")
+def get_dialplan_extensions(
+    session_asterisk: SessionAsteriskDep, user: CurrentUser
+) -> list[str]:
+    try:
+        return get_known_dialplan_extensions(session_asterisk, user)
+    except CRUDNotAllowedException as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e))

app/core/db.py

@@ -13,8 +13,9 @@
 from app.core.config import settings
 
 from app.models import *
-from app.models.asterisk import DialPlanEntry
-from app.models.crud.dialplan import Dial, Dialplan
+
+from app.telephoning.dialplan import Dialplan, Dial
+
 from app.models.user import User
 from app.models.crud.user import create_user
 from app.models.user import UserCreate, UserRole
@@ -33,9 +34,11 @@ def init_asterisk_db(session_asterisk: Session) -> None:
 
     # TODO: Figure out if we want to keep this when every SIP extension creates
     # its own dialplan (as fallback maybe?)
-    dialplan = Dialplan(session_asterisk, exten="_" + ("X" * settings.EXTENSION_DIGITS))
+    dialplan = Dialplan.from_db(
+        session_asterisk, exten="_" + ("X" * settings.EXTENSION_DIGITS)
+    )
     dialplan.add(Dial(devices=["${PJSIP_DIAL_CONTACTS(${EXTEN})}"]), prio=1)
-    dialplan.store()
+    dialplan.store(session_asterisk)
 
 
 def init_db(session: Session) -> None:

app/models/crud/asterisk.py

@@ -7,10 +7,11 @@
 
 from logging import getLogger
 from pydantic import BaseModel
-from sqlmodel import Session, delete, func, select
+from sqlmodel import Session, delete, distinct, func, select
 
 from app.core.config import settings
 from app.models.asterisk import (
+    DialPlanEntry,
     IAXFriend,
     MusicOnHold,
     PSAor,
@@ -19,7 +20,7 @@
     PSEndpoint,
 )
 from app.models.crud import CRUDNotAllowedException
-from app.models.crud.dialplan import Dial, Dialplan
+from app.telephoning.dialplan import Dial, Dialplan
 from app.models.extension import Extension
 from app.models.federation import Peer
 from app.models.user import User, UserRole
@@ -167,15 +168,15 @@ def create_or_update_callgroup(
             "You may only create callgroups with extension you've created!"
         )
 
-    plan = Dialplan(session_asterisk, extension.extension)
+    plan = Dialplan.from_db(session_asterisk, extension.extension)
     plan.add(
         Dial(
             devices=[f"${{PJSIP_DIAL_CONTACTS({e.extension})}}" for e in extensions],
             options=dialplan_options,
         ),
         1,
     )
-    plan.store()
+    plan.store(session_asterisk)
 
     logger.info(
         f"Created callgroup at {extension.extension} with participants: {participants}"
@@ -197,7 +198,7 @@ def create_iax_peer(session_asterisk: Session, peer: Peer, autocommit=True):
     try:
         session_asterisk.add(friend)
 
-        plan = Dialplan(
+        plan = Dialplan.from_db(
             session_asterisk,
             exten=f"_{peer.prefix}{'X'*peer.partner_extension_length}",
         )
@@ -209,7 +210,7 @@ def create_iax_peer(session_asterisk: Session, peer: Peer, autocommit=True):
             ),
             1,
         )
-        plan.store(autocommit)
+        plan.store(session_asterisk, autocommit)
 
         if autocommit:
             session_asterisk.commit()
@@ -237,14 +238,14 @@ def delete_iax_peer(
     if friend is None:
         raise CRUDNotAllowedException("Unkown IAX2Friend")
 
-    plan = Dialplan(
+    plan = Dialplan.from_db(
         session_asterisk,
         exten=f"_{peer.prefix}{'X'*peer.partner_extension_length}",
     )
 
     try:
         session_asterisk.delete(friend)
-        plan.delete(autocommit)
+        plan.delete(session_asterisk, autocommit)
 
         logger.info(
             f"Deleted IAX2Friend for {peer.name} and dialplan {plan.exten} from asterisk DB"
@@ -374,3 +375,12 @@ def get_extensions_with_contacts(
     extensions = list(session.exec(statement).all())
 
     return extensions
+
+
+def get_known_dialplan_extensions(session_asterisk: Session, user: User) -> list[str]:
+    if user.role != UserRole.ADMIN:
+        raise CRUDNotAllowedException("This is admin only!")
+
+    extensions = session_asterisk.exec(select(distinct(DialPlanEntry.exten))).all()
+
+    return list(extensions)