diff --git a/apps/terminal/api/__init__.py b/apps/terminal/api/__init__.py
index b6afedd51..763bae3c2 100644
--- a/apps/terminal/api/__init__.py
+++ b/apps/terminal/api/__init__.py
@@ -2,6 +2,5 @@
 #
 from .applet import *
 from .component import *
-from .db_listen_port import *
 from .session import *
 from .virtualapp import *
diff --git a/apps/terminal/api/db_listen_port.py b/apps/terminal/api/db_listen_port.py
deleted file mode 100644
index d06083768..000000000
--- a/apps/terminal/api/db_listen_port.py
+++ /dev/null
@@ -1,34 +0,0 @@
-# coding: utf-8
-#
-from rest_framework import status
-from rest_framework.decorators import action
-from rest_framework.response import Response
-from rest_framework.viewsets import GenericViewSet
-
-from assets.serializers.asset.database import DatabaseWithGatewaySerializer
-from orgs.utils import tmp_to_org
-from ..utils import db_port_manager, DBPortManager
-
-db_port_manager: DBPortManager
-
-__all__ = ['DBListenPortViewSet']
-
-
-class DBListenPortViewSet(GenericViewSet):
-    rbac_perms = {
-        '*': ['assets.view_asset'],
-    }
-    http_method_names = ['get']
-
-    def list(self, request, *args, **kwargs):
-        ports = db_port_manager.get_already_use_ports()
-        return Response(data=ports, status=status.HTTP_200_OK)
-
-    @action(methods=['get'], detail=False, url_path='db-info')
-    def db_info(self, request, *args, **kwargs):
-        port = request.query_params.get("port")
-        db = db_port_manager.get_db_by_port(port)
-
-        with tmp_to_org(db.org):
-            serializer = DatabaseWithGatewaySerializer(instance=db)
-            return Response(data=serializer.data, status=status.HTTP_200_OK)
diff --git a/apps/terminal/migrations/0006_endpoint_oracle_port.py b/apps/terminal/migrations/0006_endpoint_oracle_port.py
new file mode 100644
index 000000000..f79db16ec
--- /dev/null
+++ b/apps/terminal/migrations/0006_endpoint_oracle_port.py
@@ -0,0 +1,20 @@
+# Generated by Django 4.1.13 on 2025-04-01 07:44
+
+import common.db.fields
+import django.core.validators
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('terminal', '0005_endpoint_vnc_port'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='endpoint',
+            name='oracle_port',
+            field=common.db.fields.PortField(default=15210, validators=[django.core.validators.MinValueValidator(0), django.core.validators.MaxValueValidator(65535)], verbose_name='Oracle port'),
+        ),
+    ]
diff --git a/apps/terminal/models/component/endpoint.py b/apps/terminal/models/component/endpoint.py
index e0ff92650..c4b44801d 100644
--- a/apps/terminal/models/component/endpoint.py
+++ b/apps/terminal/models/component/endpoint.py
@@ -22,6 +22,7 @@ class Endpoint(JMSBaseModel):
     postgresql_port = PortField(default=54320, verbose_name=_('PostgreSQL port'))
     redis_port = PortField(default=63790, verbose_name=_('Redis port'))
     sqlserver_port = PortField(default=14330, verbose_name=_('SQLServer port'))
+    oracle_port = PortField(default=15210,verbose_name=_('Oracle port'))
     vnc_port = PortField(default=15900, verbose_name=_('VNC port'))
 
     comment = models.TextField(default='', blank=True, verbose_name=_('Comment'))
@@ -37,17 +38,10 @@ class Endpoint(JMSBaseModel):
         return self.name
 
     def get_port(self, target_instance, protocol):
-        from terminal.utils import db_port_manager
-        from assets.const import DatabaseTypes, Protocol
-
-        if isinstance(target_instance, Asset) and \
-                target_instance.is_type(DatabaseTypes.ORACLE) and \
-                protocol == Protocol.oracle:
-            port = db_port_manager.get_port_by_db(target_instance)
-        else:
-            if protocol in [Protocol.sftp, Protocol.telnet]:
-                protocol = Protocol.ssh
-            port = getattr(self, f'{protocol}_port', 0)
+        from assets.const import Protocol
+        if protocol in [Protocol.sftp, Protocol.telnet]:
+            protocol = Protocol.ssh
+        port = getattr(self, f'{protocol}_port', 0)
         return port
 
     def is_default(self):
diff --git a/apps/terminal/serializers/endpoint.py b/apps/terminal/serializers/endpoint.py
index 46c1f343e..effd99ca1 100644
--- a/apps/terminal/serializers/endpoint.py
+++ b/apps/terminal/serializers/endpoint.py
@@ -5,30 +5,18 @@ from acls.serializers.rules import ip_group_child_validator, ip_group_help_text
 from common.serializers import BulkModelSerializer
 from common.serializers.fields import ObjectRelatedField
 from ..models import Endpoint, EndpointRule
-from ..utils import db_port_manager
 
 __all__ = ['EndpointSerializer', 'EndpointRuleSerializer']
 
 
 class EndpointSerializer(BulkModelSerializer):
-    # 解决 luna 处理繁琐的问题, 返回 magnus 监听的当前 db 的 port
-    oracle_port = serializers.SerializerMethodField(label=_('Oracle port'))
-    oracle_port_range = serializers.CharField(
-        max_length=128, default=db_port_manager.oracle_port_range, read_only=True,
-        label=_('Oracle port range'),
-        help_text=_(
-            'Oracle proxy server listen port is dynamic, Each additional Oracle '
-            'database instance adds a port listener'
-        )
-    )
-
     class Meta:
         model = Endpoint
         fields_mini = ['id', 'name']
         fields_small = [
             'host', 'https_port', 'http_port', 'ssh_port', 'rdp_port',
             'mysql_port', 'mariadb_port', 'postgresql_port', 'redis_port', 'vnc_port',
-            'oracle_port_range', 'oracle_port', 'sqlserver_port', 'is_active'
+            'oracle_port', 'sqlserver_port', 'is_active'
         ]
         fields = fields_mini + fields_small + [
             'comment', 'date_created', 'date_updated', 'created_by'
@@ -41,13 +29,6 @@ class EndpointSerializer(BulkModelSerializer):
             )
             },
         }
-
-    def get_oracle_port(self, obj: Endpoint):
-        view = self.context.get('view')
-        if not view or view.action not in ['smart']:
-            return 0
-        return obj.get_port(view.target_instance, view.target_protocol)
-
     def get_extra_kwargs(self):
         extra_kwargs = super().get_extra_kwargs()
         model_fields = self.Meta.model._meta.fields
diff --git a/apps/terminal/signal_handlers/__init__.py b/apps/terminal/signal_handlers/__init__.py
index 08a6ac407..503db0da3 100644
--- a/apps/terminal/signal_handlers/__init__.py
+++ b/apps/terminal/signal_handlers/__init__.py
@@ -1,5 +1,4 @@
 from .applet import *
-from .db_port import *
 from .session import *
 from .session_sharing import *
 from .terminal import *
diff --git a/apps/terminal/signal_handlers/applet.py b/apps/terminal/signal_handlers/applet.py
index 99eb2fca7..0fa622cec 100644
--- a/apps/terminal/signal_handlers/applet.py
+++ b/apps/terminal/signal_handlers/applet.py
@@ -12,9 +12,7 @@ from orgs.utils import tmp_to_builtin_org
 from users.models import User
 from ..models import Applet, AppletHost
 from ..tasks import applet_host_generate_accounts
-from ..utils import DBPortManager
 
-db_port_manager: DBPortManager
 logger = get_logger(__file__)
 
 
diff --git a/apps/terminal/signal_handlers/db_port.py b/apps/terminal/signal_handlers/db_port.py
deleted file mode 100644
index 3ba3d9225..000000000
--- a/apps/terminal/signal_handlers/db_port.py
+++ /dev/null
@@ -1,38 +0,0 @@
-from django.db.models.signals import post_delete, post_save
-from django.dispatch import receiver
-
-from assets.models import Asset, Database
-from common.decorators import on_transaction_commit
-from common.signals import django_ready
-from common.utils import get_logger
-from ..utils import db_port_manager
-
-logger = get_logger(__file__)
-
-
-@receiver(django_ready)
-def check_db_port_mapper(sender, **kwargs):
-    logger.info('Check oracle ports (MAGNUS_ORACLE_PORTS)')
-    try:
-        db_port_manager.check()
-    except Exception as e:
-        # 新部署会显示 assets_database 表不存在
-        logger.warning('(Ignore) {}'.format(e))
-
-
-@receiver(post_save, sender=Database)
-def on_db_created(sender, instance: Database, created, **kwargs):
-    if instance.type != 'oracle':
-        return
-    if not created:
-        return
-    logger.info("Oracle create signal recv: {} {}".format(instance, instance.type))
-    db_port_manager.check()
-
-
-@receiver(post_delete, sender=Database)
-def on_db_delete(sender, instance, **kwargs):
-    if instance.type != 'oracle':
-        return
-    logger.info("Oracle delete signal recv: {}".format(instance))
-    db_port_manager.check()
diff --git a/apps/terminal/signal_handlers/terminal.py b/apps/terminal/signal_handlers/terminal.py
index a227c0cf1..2dd698576 100644
--- a/apps/terminal/signal_handlers/terminal.py
+++ b/apps/terminal/signal_handlers/terminal.py
@@ -7,9 +7,6 @@ from common.utils import get_logger
 from common.utils.connection import RedisPubSub
 from ..const import TaskNameType
 from ..models import Task, Session
-from ..utils import DBPortManager
-
-db_port_manager: DBPortManager
 
 logger = get_logger(__file__)
 
diff --git a/apps/terminal/urls/api_urls.py b/apps/terminal/urls/api_urls.py
index e941acd21..fbbe5ca8f 100644
--- a/apps/terminal/urls/api_urls.py
+++ b/apps/terminal/urls/api_urls.py
@@ -29,7 +29,6 @@ router.register(r'applet-hosts/((?P<host>[^/.]+)/)?applets', api.AppletHostApple
 router.register(r'applet-hosts', api.AppletHostViewSet, 'applet-host')
 router.register(r'applet-publications', api.AppletPublicationViewSet, 'applet-publication')
 router.register(r'applet-host-deployments', api.AppletHostDeploymentViewSet, 'applet-host-deployment')
-router.register(r'db-listen-ports', api.DBListenPortViewSet, 'db-listen-ports')
 router.register(r'virtual-apps', api.VirtualAppViewSet, 'virtual-app')
 router.register(r'app-providers', api.AppProviderViewSet, 'app-provider')
 router.register(r'app-providers/((?P<provider>[^/.]+)/)?apps', api.AppProviderAppViewSet, 'app-provider-app')
diff --git a/apps/terminal/utils/__init__.py b/apps/terminal/utils/__init__.py
index 0110ad579..930690d2d 100644
--- a/apps/terminal/utils/__init__.py
+++ b/apps/terminal/utils/__init__.py
@@ -1,3 +1,2 @@
 from .components import *
 from .common import *
-from .db_port_mapper import *
diff --git a/apps/terminal/utils/db_port_mapper.py b/apps/terminal/utils/db_port_mapper.py
deleted file mode 100644
index 796632a65..000000000
--- a/apps/terminal/utils/db_port_mapper.py
+++ /dev/null
@@ -1,156 +0,0 @@
-from django.conf import settings
-from django.core.cache import cache
-from django.utils.translation import gettext_lazy as _
-
-from assets.const import DatabaseTypes
-from assets.models import Database
-from common.decorators import Singleton
-from common.exceptions import JMSException
-from common.utils import get_logger, get_object_or_none
-from orgs.utils import tmp_to_root_org
-
-logger = get_logger(__file__)
-
-
-@Singleton
-class DBPortManager:
-    """ 管理端口-数据库ID的映射, Magnus 要使用 """
-    CACHE_KEY = 'PORT_DB_MAPPER'
-
-    def __init__(self):
-        oracle_ports = self.oracle_port_range
-        try:
-            port_start, port_end = oracle_ports.split('-')
-            port_start, port_end = int(port_start), int(port_end)
-        except Exception as e:
-            logger.error('MAGNUS_ORACLE_PORTS config error: {}'.format(e))
-            port_start, port_end = 30000, 30100
-
-        self.port_start, self.port_end = port_start, port_end
-        # 可以使用的端口列表
-        self.all_avail_ports = list(range(self.port_start, self.port_end + 1))
-
-    @property
-    def oracle_port_range(self):
-        oracle_ports = settings.MAGNUS_ORACLE_PORTS
-        if not oracle_ports and settings.MAGNUS_PORTS:
-            oracle_ports = settings.MAGNUS_PORTS
-        return oracle_ports
-
-    @staticmethod
-    def fetch_dbs():
-        with tmp_to_root_org():
-            dbs = Database.objects.filter(platform__type=DatabaseTypes.ORACLE).order_by('id')
-            return dbs
-
-    def check(self):
-        dbs = self.fetch_dbs()
-        mapper = self.get_mapper()
-        db_ids = [str(db.id) for db in dbs]
-        db_ids_to_add = list(set(db_ids) - set(mapper.values()))
-        mapper = self.bulk_add(db_ids_to_add, mapper)
-
-        db_ids_to_pop = set(mapper.values()) - set(db_ids)
-        mapper = self.bulk_pop(db_ids_to_pop, mapper)
-
-        if db_ids_to_add or db_ids_to_pop:
-            self.set_mapper(mapper)
-
-        if settings.DEBUG:
-            logger.debug("Oracle listen ports: {}".format(len(mapper.keys())))
-
-    def init(self):
-        dbs = self.fetch_dbs()
-        db_ids = dbs.values_list('id', flat=True)
-        db_ids = [str(i) for i in db_ids]
-        mapper = dict(zip(self.all_avail_ports, list(db_ids)))
-        self.set_mapper(mapper)
-        return mapper
-
-    def bulk_add(self, db_ids, mapper):
-        for db_id in db_ids:
-            avail_port = self.get_next_avail_port(mapper)
-            mapper[avail_port] = str(db_id)
-        return mapper
-
-    def bulk_pop(self, db_ids, mapper):
-        new_mapper = {port: str(db_id) for port, db_id in mapper.items() if db_id not in db_ids}
-        return new_mapper
-
-    def get_port_by_db(self, db, raise_exception=True):
-        mapper = self.get_mapper()
-        for port, db_id in mapper.items():
-            if db_id == str(db.id):
-                return port
-
-        if raise_exception:
-            error = _(
-                'No available port is matched. '
-                'The number of databases may have exceeded the number of ports '
-                'open to the database agent service, '
-                'Contact the administrator to open more ports.'
-            )
-            raise JMSException(error)
-
-    def get_db_by_port(self, port):
-        try:
-            port = int(port)
-        except Exception as e:
-            raise JMSException('Port type error: {}'.format(e))
-        mapper = self.get_mapper()
-        db_id = mapper.get(port, None)
-        if not db_id:
-            raise JMSException('Database not in port-db mapper, port: {}'.format(port))
-        with tmp_to_root_org():
-            db = get_object_or_none(Database, id=db_id)
-        if not db:
-            raise JMSException('Database not exists, db id: {}'.format(db_id))
-        return db
-
-    def get_next_avail_port(self, mapper=None):
-        if mapper is None:
-            mapper = self.get_mapper()
-        already_use_ports = [int(i) for i in mapper.keys()]
-        avail_ports = sorted(list(set(self.all_avail_ports) - set(already_use_ports)))
-        if len(avail_ports) <= 0:
-            msg = _('No ports can be used, check and modify the limit on the number '
-                    'of ports that Magnus listens on in the configuration file.')
-            tips = _('All available port count: {}, Already use port count: {}').format(
-                len(self.all_avail_ports), len(already_use_ports)
-            )
-            error = msg + tips
-            raise JMSException(error)
-        port = avail_ports[0]
-        logger.debug('Get next available port: {}'.format(port))
-        return port
-
-    def get_already_use_ports(self):
-        mapper = self.get_mapper()
-        return sorted([int(i) for i in mapper.keys()])
-
-    @staticmethod
-    def oracle_ports_setting_changed():
-        oracle_ports_cache = cache.get('MAGNUS_ORACLE_PORTS') or ''
-        if settings.MAGNUS_ORACLE_PORTS.split('-')[0] != oracle_ports_cache.split('-')[0]:
-            logger.info('Oracle ports setting changed')
-            return True
-        return False
-
-    def get_mapper(self):
-        mapper = cache.get(self.CACHE_KEY, {})
-        if not mapper or self.oracle_ports_setting_changed():
-            # redis 可能被清空，重新初始化一下
-            mapper = self.init()
-        return mapper
-
-    def set_mapper(self, value):
-        """
-        value: {
-            port: db_id
-        }
-        """
-        cache.set(self.CACHE_KEY, value, timeout=None)
-        cache.set('MAGNUS_ORACLE_PORTS', settings.MAGNUS_ORACLE_PORTS)
-
-
-db_port_manager = DBPortManager()