Code Duplication - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 86-95 lines in 2 locations

myems-api/core/combinedequipment.py 1 location


        _ = req
        resp.status = falcon.HTTP_200

    @staticmethod
    def on_get(req, resp):
        """
        Handle GET requests to retrieve all combined equipments

        Returns a list of all combined equipments with their metadata including:
        - Equipment ID, name, and UUID
        - Input/output counting status
        - Cost center information
        - SVG diagram reference
        - Camera URL for monitoring
        - Description and QR code

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        # Check authentication method (API key or session)
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)

        search_query = req.get_param('q', default=None)
        if search_query is not None:
            search_query = search_query.strip()
        else:
            search_query = ''

        # Connect to database and retrieve cost centers
        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        # Query to retrieve all cost centers for reference
        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        # Build cost center dictionary for quick lookup
        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        # Query to retrieve all SVG diagrams for reference
        query = (" SELECT id, name, uuid "
                 " FROM tbl_svgs ")
        cursor.execute(query)
        rows_svgs = cursor.fetchall()

        # Build SVG dictionary for quick lookup
        svg_dict = dict()
        if rows_svgs is not None and len(rows_svgs) > 0:
            for row in rows_svgs:
                svg_dict[row[0]] = {"id": row[0],
                                    "name": row[1],
                                    "uuid": row[2]}

        # Query to retrieve all combined equipments
        query = (" SELECT id, name, uuid, "
                 "        is_input_counted, is_output_counted, "
                 "        cost_center_id, svg_id, camera_url, description "
                 " FROM tbl_combined_equipments ")
        params = []
        if search_query:
            query += " WHERE name LIKE %s OR   description LIKE %s "
            params = [f'%{search_query}%', f'%{search_query}%']
        query += " ORDER BY id "
        cursor.execute(query, params)
        rows_combined_equipments = cursor.fetchall()

        # Build result list with all combined equipment data
        result = list()
        if rows_combined_equipments is not None and len(rows_combined_equipments) > 0:
            for row in rows_combined_equipments:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "is_input_counted": bool(row[3]),
                               "is_output_counted": bool(row[4]),
                               "cost_center": cost_center_dict.get(row[5], None),
                               "svg": svg_dict.get(row[6], None),
                               "camera_url": row[7],
                               "description": row[8],
                               "qrcode": 'combinedequipment:' + row[2]}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

myems-api/core/shopfloor.py 1 location


        resp.status = falcon.HTTP_200
        _ = req

    @staticmethod
    def on_get(req, resp):
        """
        Handle GET requests to retrieve all shopfloors

        Returns a list of all shopfloors with their complete information including:
        - Shopfloor ID, name, and UUID
        - Associated contact and cost center information
        - Shopfloor specifications and parameters
        - Related equipment, sensors, and meter associations
        - Working calendar and command configurations

        Args:
            req: Falcon request object
            resp: Falcon response object
        """
        # Check authentication method (API key or session)
        if 'API-KEY' not in req.headers or \
                not isinstance(req.headers['API-KEY'], str) or \
                len(str.strip(req.headers['API-KEY'])) == 0:
            access_control(req)
        else:
            api_key_control(req)

        search_query = req.get_param('q', default=None)
        if search_query is not None and len(search_query.strip()) > 0:
            search_query = search_query.strip()
        else:
            search_query = ''

        cnx = mysql.connector.connect(**config.myems_system_db)
        cursor = cnx.cursor()

        query = (" SELECT id, name, uuid "
                 " FROM tbl_contacts ")
        cursor.execute(query)
        rows_contacts = cursor.fetchall()

        contact_dict = dict()
        if rows_contacts is not None and len(rows_contacts) > 0:
            for row in rows_contacts:
                contact_dict[row[0]] = {"id": row[0],
                                        "name": row[1],
                                        "uuid": row[2]}

        query = (" SELECT id, name, uuid "
                 " FROM tbl_cost_centers ")
        cursor.execute(query)
        rows_cost_centers = cursor.fetchall()

        cost_center_dict = dict()
        if rows_cost_centers is not None and len(rows_cost_centers) > 0:
            for row in rows_cost_centers:
                cost_center_dict[row[0]] = {"id": row[0],
                                            "name": row[1],
                                            "uuid": row[2]}

        query = (" SELECT id, name, uuid, "
                 "        area, is_input_counted, "
                 "        contact_id, cost_center_id, description "
                 " FROM tbl_shopfloors ")
        params = []
        if search_query:
            query += " WHERE name LIKE %s OR description LIKE %s "
            params = [f'%{search_query}%', f'%{search_query}%']
        query += " ORDER BY id "
        cursor.execute(query, params)
        rows_shopfloors = cursor.fetchall()

        result = list()
        if rows_shopfloors is not None and len(rows_shopfloors) > 0:
            for row in rows_shopfloors:
                meta_result = {"id": row[0],
                               "name": row[1],
                               "uuid": row[2],
                               "area": row[3],
                               "is_input_counted": bool(row[4]),
                               "contact": contact_dict.get(row[5], None),
                               "cost_center": cost_center_dict.get(row[6], None),
                               "description": row[7],
                               "qrcode": "shopfloor:" + row[2]}
                result.append(meta_result)

        cursor.close()
        cnx.close()
        resp.text = json.dumps(result)

    @staticmethod
    @user_logger

		@@ 35-129 (lines=95) @@
32		_ = req
33		resp.status = falcon.HTTP_200
34
35		@staticmethod
36		def on_get(req, resp):
37		"""
38		Handle GET requests to retrieve all combined equipments
39
40		Returns a list of all combined equipments with their metadata including:
41		- Equipment ID, name, and UUID
42		- Input/output counting status
43		- Cost center information
44		- SVG diagram reference
45		- Camera URL for monitoring
46		- Description and QR code
47
48		Args:
49		req: Falcon request object
50		resp: Falcon response object
51		"""
52		# Check authentication method (API key or session)
53		if 'API-KEY' not in req.headers or \
54		not isinstance(req.headers['API-KEY'], str) or \
55		len(str.strip(req.headers['API-KEY'])) == 0:
56		access_control(req)
57		else:
58		api_key_control(req)
59
60		search_query = req.get_param('q', default=None)
61		if search_query is not None:
62		search_query = search_query.strip()
63		else:
64		search_query = ''
65
66		# Connect to database and retrieve cost centers
67		cnx = mysql.connector.connect(**config.myems_system_db)
68		cursor = cnx.cursor()
69
70		# Query to retrieve all cost centers for reference
71		query = (" SELECT id, name, uuid "
72		" FROM tbl_cost_centers ")
73		cursor.execute(query)
74		rows_cost_centers = cursor.fetchall()
75
76		# Build cost center dictionary for quick lookup
77		cost_center_dict = dict()
78		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
79		for row in rows_cost_centers:
80		cost_center_dict[row[0]] = {"id": row[0],
81		"name": row[1],
82		"uuid": row[2]}
83
84		# Query to retrieve all SVG diagrams for reference
85		query = (" SELECT id, name, uuid "
86		" FROM tbl_svgs ")
87		cursor.execute(query)
88		rows_svgs = cursor.fetchall()
89
90		# Build SVG dictionary for quick lookup
91		svg_dict = dict()
92		if rows_svgs is not None and len(rows_svgs) > 0:
93		for row in rows_svgs:
94		svg_dict[row[0]] = {"id": row[0],
95		"name": row[1],
96		"uuid": row[2]}
97
98		# Query to retrieve all combined equipments
99		query = (" SELECT id, name, uuid, "
100		" is_input_counted, is_output_counted, "
101		" cost_center_id, svg_id, camera_url, description "
102		" FROM tbl_combined_equipments ")
103		params = []
104		if search_query:
105		query += " WHERE name LIKE %s OR description LIKE %s "
106		params = [f'%{search_query}%', f'%{search_query}%']
107		query += " ORDER BY id "
108		cursor.execute(query, params)
109		rows_combined_equipments = cursor.fetchall()
110
111		# Build result list with all combined equipment data
112		result = list()
113		if rows_combined_equipments is not None and len(rows_combined_equipments) > 0:
114		for row in rows_combined_equipments:
115		meta_result = {"id": row[0],
116		"name": row[1],
117		"uuid": row[2],
118		"is_input_counted": bool(row[3]),
119		"is_output_counted": bool(row[4]),
120		"cost_center": cost_center_dict.get(row[5], None),
121		"svg": svg_dict.get(row[6], None),
122		"camera_url": row[7],
123		"description": row[8],
124		"qrcode": 'combinedequipment:' + row[2]}
125		result.append(meta_result)
126
127		cursor.close()
128		cnx.close()
129		resp.text = json.dumps(result)
130
131		@staticmethod
132		@user_logger

		@@ 35-120 (lines=86) @@
32		resp.status = falcon.HTTP_200
33		_ = req
34
35		@staticmethod
36		def on_get(req, resp):
37		"""
38		Handle GET requests to retrieve all shopfloors
39
40		Returns a list of all shopfloors with their complete information including:
41		- Shopfloor ID, name, and UUID
42		- Associated contact and cost center information
43		- Shopfloor specifications and parameters
44		- Related equipment, sensors, and meter associations
45		- Working calendar and command configurations
46
47		Args:
48		req: Falcon request object
49		resp: Falcon response object
50		"""
51		# Check authentication method (API key or session)
52		if 'API-KEY' not in req.headers or \
53		not isinstance(req.headers['API-KEY'], str) or \
54		len(str.strip(req.headers['API-KEY'])) == 0:
55		access_control(req)
56		else:
57		api_key_control(req)
58
59		search_query = req.get_param('q', default=None)
60		if search_query is not None and len(search_query.strip()) > 0:
61		search_query = search_query.strip()
62		else:
63		search_query = ''
64
65		cnx = mysql.connector.connect(**config.myems_system_db)
66		cursor = cnx.cursor()
67
68		query = (" SELECT id, name, uuid "
69		" FROM tbl_contacts ")
70		cursor.execute(query)
71		rows_contacts = cursor.fetchall()
72
73		contact_dict = dict()
74		if rows_contacts is not None and len(rows_contacts) > 0:
75		for row in rows_contacts:
76		contact_dict[row[0]] = {"id": row[0],
77		"name": row[1],
78		"uuid": row[2]}
79
80		query = (" SELECT id, name, uuid "
81		" FROM tbl_cost_centers ")
82		cursor.execute(query)
83		rows_cost_centers = cursor.fetchall()
84
85		cost_center_dict = dict()
86		if rows_cost_centers is not None and len(rows_cost_centers) > 0:
87		for row in rows_cost_centers:
88		cost_center_dict[row[0]] = {"id": row[0],
89		"name": row[1],
90		"uuid": row[2]}
91
92		query = (" SELECT id, name, uuid, "
93		" area, is_input_counted, "
94		" contact_id, cost_center_id, description "
95		" FROM tbl_shopfloors ")
96		params = []
97		if search_query:
98		query += " WHERE name LIKE %s OR description LIKE %s "
99		params = [f'%{search_query}%', f'%{search_query}%']
100		query += " ORDER BY id "
101		cursor.execute(query, params)
102		rows_shopfloors = cursor.fetchall()
103
104		result = list()
105		if rows_shopfloors is not None and len(rows_shopfloors) > 0:
106		for row in rows_shopfloors:
107		meta_result = {"id": row[0],
108		"name": row[1],
109		"uuid": row[2],
110		"area": row[3],
111		"is_input_counted": bool(row[4]),
112		"contact": contact_dict.get(row[5], None),
113		"cost_center": cost_center_dict.get(row[6], None),
114		"description": row[7],
115		"qrcode": "shopfloor:" + row[2]}
116		result.append(meta_result)
117
118		cursor.close()
119		cnx.close()
120		resp.text = json.dumps(result)
121
122		@staticmethod
123		@user_logger

MyEMS / myems

Code Duplication Length = 86-95 lines in 2 locations

myems-api/core/combinedequipment.py 1 location

myems-api/core/shopfloor.py 1 location