ags_validator.py

# -*- coding: utf-8 -*-

"""
/***************************************************************************
 AGS2DB
								 A QGIS plugin
 This plugin parses an AGS file and creates a database from it
 Generated by Plugin Builder: http://g-sherman.github.io/Qgis-Plugin-Builder/
							  -------------------
		begin                : 2023-04-19
		copyright            : (C) 2023 by Oliver Burdekin / burdGIS
		email                : info@burdgis.com
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
"""

__author__ = 'Oliver Burdekin / burdGIS'
__date__ = '2023-04-19'
__copyright__ = '(C) 2023 by Oliver Burdekin / burdGIS'

# This will get replaced with a git SHA1 when you do a git archive

__revision__ = '$Format:%H$'

from qgis.PyQt.QtCore import QCoreApplication, QUrl, QEventLoop, QTimer
from qgis.PyQt.QtNetwork import QNetworkRequest, QNetworkReply, QHttpMultiPart, QHttpPart
from qgis.core import (
	QgsProcessingAlgorithm,
	QgsProcessingParameterFile,
	QgsProcessingParameterFileDestination,
	QgsProcessingParameterEnum,
	QgsNetworkAccessManager,
	# QgsApplication,  # uncomment if you use QGIS auth manager
)
import os

class AGSValidatorAlgorithm(QgsProcessingAlgorithm):
	"""
	This is the algorithm class for the AGS validator.
	"""

	# Define constants for the algorithm's parameters
	INPUT = 'INPUT'
	DICTIONARY = 'DICTIONARY'
	CHECKERS = 'CHECKERS'
	OUTPUT = 'OUTPUT'
	
	# Define constants for the dictionary choices and checker options
	DICTIONARY_OPTIONS = ['None', '4.0.3', '4.0.4', '4.1', '4.1.1']  
	DICTIONARY_ALIASES = {'None': 'v4_1_1','4.0.3': 'v4_0_3','4.0.4': 'v4_0_4','4.1': 'v4_1','4.1.1': 'v4_1_1'}
	CHECKER_OPTIONS = ['ags', 'bgs']  

	def initAlgorithm(self, config):
		"""
		Here we define the inputs and output of the algorithm, along
		with some other properties.
		"""

		# Define the inputs for the algorithm
		self.addParameter(
			QgsProcessingParameterFile(
				self.INPUT,
				self.tr("Input File"),
				behavior=QgsProcessingParameterFile.Behavior.File,
				fileFilter="All files (*.*)"
			)
		)
		self.addParameter(
			QgsProcessingParameterEnum(
				self.DICTIONARY,
				self.tr("AGS version"),
				options=self.DICTIONARY_OPTIONS,
				defaultValue=0
			)
		)

		self.addParameter(
			QgsProcessingParameterEnum(
				self.CHECKERS,
				self.tr("Checkers"),
				options=self.CHECKER_OPTIONS,
				allowMultiple=True,
				defaultValue=0
			)
		)

		# Define the outputs for the algorithm
		self.addParameter(
			QgsProcessingParameterFileDestination(
				self.OUTPUT,
				self.tr("Output File"),
				fileFilter="Text files (*.txt)"
			)
		)

   
	def processAlgorithm(self, parameters, context, feedback):
		# Params
		file_path = self.parameterAsFile(parameters, self.INPUT, context)
		dictionary = self.DICTIONARY_OPTIONS[self.parameterAsEnum(parameters, self.DICTIONARY, context)]
		dictionary_alias = self.DICTIONARY_ALIASES[dictionary]
		checkers_selected_indices = self.parameterAsEnums(parameters, self.CHECKERS, context)
		checkers_selected = [self.CHECKER_OPTIONS[i] for i in checkers_selected_indices]
		output_file = self.parameterAsFileOutput(parameters, self.OUTPUT, context)

		file_name = os.path.basename(file_path)
		url = 'https://agsapi.bgs.ac.uk/v6/validate/'
		fmt = 'text'

		# Read file
		with open(file_path, 'rb') as f:
			file_content = f.read()

		# --- Build multipart/form-data via Qt ---

		mp = QHttpMultiPart(QHttpMultiPart.ContentType.FormDataType)

		# std_dictionary
		p_std = QHttpPart()
		p_std.setHeader(QNetworkRequest.KnownHeaders.ContentDispositionHeader, 'form-data; name="std_dictionary"')
		p_std.setBody(dictionary_alias.encode('utf-8'))
		mp.append(p_std)

		# checkers (repeat field)
		for c in checkers_selected:
			p = QHttpPart()
			p.setHeader(QNetworkRequest.KnownHeaders.ContentDispositionHeader, 'form-data; name="checkers"')
			p.setBody(str(c).encode('utf-8'))
			mp.append(p)

		# fmt
		p_fmt = QHttpPart()
		p_fmt.setHeader(QNetworkRequest.KnownHeaders.ContentDispositionHeader, 'form-data; name="fmt"')
		p_fmt.setBody(fmt.encode('utf-8'))
		mp.append(p_fmt)

		# file
		p_file = QHttpPart()
		p_file.setHeader(QNetworkRequest.KnownHeaders.ContentDispositionHeader,
						f'form-data; name="files"; filename="{file_name}"')
		p_file.setHeader(QNetworkRequest.KnownHeaders.ContentTypeHeader, 'text/plain')  # or 'application/octet-stream'
		p_file.setBody(file_content)
		mp.append(p_file)

		# POST via QGIS network manager (proxy/auth aware)
		req = QNetworkRequest(QUrl(url))
		# If you use a QGIS auth config, uncomment:
		# from qgis.core import QgsApplication
		# QgsApplication.authManager().updateNetworkRequest(req, "your_authcfg_id")

		nam = QgsNetworkAccessManager.instance()
		reply = nam.post(req, mp)
		mp.setParent(reply)  # keep multipart alive until finished

		# Block until finished (with a 60s safety timeout)
		loop = QEventLoop()
		timer = QTimer()
		timer.setSingleShot(True)
		timer.timeout.connect(lambda: reply.abort())
		reply.finished.connect(loop.quit)
		timer.start(60000)  # adjust or remove if you prefer no timeout
		loop.exec()
		timer.stop()

		# Error handling
		if reply.error() != QNetworkReply.NetworkError.NoError:
			err = reply.errorString()
			reply.deleteLater()
			feedback.reportError(f"Error calling API: {err}")
			raise Exception(f"Error calling API: {err}")

		# Success
		data = bytes(reply.readAll()).decode('utf-8', errors='replace')
		reply.deleteLater()

		with open(output_file, 'w', encoding='utf-8') as f:
			f.write(data)

		feedback.pushInfo('API call was successful.')
		return {self.OUTPUT: output_file}


	
	def processing_log(self, message):
		"""
		Logs a message to the Processing log.
		"""
		self.logMessage(message)

	def name(self):
		"""
		Returns the algorithm name, used for identifying the algorithm. This
		string should be fixed for the algorithm, and must not be localised.
		The name should be unique within each provider. Names should contain
		lowercase alphanumeric characters only and no spaces or other
		formatting characters.
		"""
		return 'AGS validator'

	def displayName(self):
		"""
		Returns the translated algorithm name, which should be used for any
		user-visible display of the algorithm name.
		"""
		return self.tr(self.name())

	def group(self):
		"""
		Returns the name of the group this algorithm belongs to. This string
		should be localised.
		"""
		return self.tr(self.groupId())

	def groupId(self):
		"""
		Returns the unique ID of the group this algorithm belongs to. This
		string should be fixed for the algorithm, and must not be localised.
		The group id should be unique within each provider. Group id should
		contain lowercase alphanumeric characters only and no spaces or other
		formatting characters.
		"""
		return ''

	def tr(self, string):
		return QCoreApplication.translate('Processing', string)

	def createInstance(self):
		return AGSValidatorAlgorithm()