audrius/pms
1
0
Fork 1
mirror of https://github.com/OCA/pms.git synced 2025-01-29 00:17:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[IMP]pms_ocr_klippa: improvement mapped data and partial results

Browse Source
This commit is contained in:
Darío Lodeiros
2024-05-22 19:03:12 +02:00
parent b2e3d4a663
commit ab3fd49713
3 changed files with 210 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
pms_ocr_klippa/models/klippa_log.py
View File

@@ -29,8 +29,8 @@ class KlippaLog(models.Model):
help="Response",
)
klippa_status = fields.Char(
string="Status",
help="Status",
string="Klippa Status",
help="Klippa Status",
)
request_datetime = fields.Datetime(
string="Request Date",
@@ -84,6 +84,14 @@ class KlippaLog(models.Model):
string="Error",
help="Error",
)
nominatim_status = fields.Char(
string="Nominatim Status",
help="Nominatim Status",
)
nominatim_response = fields.Text(
string="Nominatim Response",
help="Nominatim Response",
)
def clean_log_data(self, offset=60):
"""Clean log data older than the offset.

296
pms_ocr_klippa/models/pms_property.py
View File

@@ -12,6 +12,25 @@ _logger = logging.getLogger(__name__)
NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
CHECKIN_FIELDS = {
"nationality": "partner_id.nationality_id.id",
"country_id": "partner_id.residence_country_id.id",
"firstname": "partner_id.firstname",
"lastname": "partner_id.lastname",
"lastname2": "partner_id.lastname2",
"gender": "partner_id.gender",
"birthdate": "partner_id.birthdate_date",
"document_type": "document_type_id.id",
"document_expedition_date": "document_expedition_date",
"document_support_number": "document_support_number",
"document_number": "name",
"residence_street": "partner_id.residence_street",
"residence_city": "partner_id.residence_city",
"country_state": "partner_id.residence_state_id.id",
"document_country_id": "document_country_id",
"zip": "partner_id.zip",
}
class PmsProperty(models.Model):
_inherit = "pms.property"
@@ -78,7 +97,24 @@ class PmsProperty(models.Model):
raise ValidationError(_("Error calling Klippa OCR API"))
document_data = json_data["data"]["parsed"]
init_mapped_datetime = datetime.now()
mapped_data = self._map_klippa_data(document_data)
if mapped_data.get("nominatim_status"):
log_data.update(
{
"nominatim_status": mapped_data["nominatim_status"],
}
)
mapped_data.pop("nominatim_status")
if mapped_data.get("nominatim_response"):
log_data.update(
{
"nominatim_response": mapped_data["nominatim_response"],
}
)
mapped_data.pop("nominatim_response")
log_data.update(
{
"service_response": mapped_data,
@@ -107,18 +143,7 @@ class PmsProperty(models.Model):
def _map_klippa_data(self, document_data):
mapped_data = {}
found_partner = False
if document_data.get("personal_number", False):
found_partner = (
self.env["res.partner.id_number"]
.search(
[
("name", "=", document_data["personal_number"]["value"]),
],
limit=1,
)
.partner_id
)
key_document_number, key_personal_number = self._get_number_keys(document_data)
for key, dict_value in document_data.items():
if dict_value and isinstance(dict_value, dict):
value = dict_value.get("value", False)
@@ -131,8 +156,6 @@ class PmsProperty(models.Model):
# Document Data --------------------------------------------------
elif key == "issuing_country" and value:
mapped_data["document_country_id"] = self._get_country_id(value)
elif key == "document_number" and value:
mapped_data["document_support_number"] = value
elif key == "document_type" and value:
mapped_data["document_type"] = self._get_document_type(
klippa_type=value,
@@ -141,9 +164,11 @@ class PmsProperty(models.Model):
if document_data.get("issuing_country")
else False
),
)
).id
elif key == "personal_number" and value:
mapped_data["document_number"] = value
mapped_data[key_personal_number] = value
elif key == "document_number" and value:
mapped_data[key_document_number] = value
elif key == "date_of_issue" and value:
mapped_data["document_expedition_date"] = datetime.strptime(
value, "%Y-%m-%dT%H:%M:%S"
@@ -190,6 +215,21 @@ class PmsProperty(models.Model):
).date()
elif key == "nationality" and value:
mapped_data["nationality"] = self._get_country_id(value)
# If the document number exist and not get the complete checkin information
# recovery the lost data from the found document
if mapped_data.get("document_number") and not all(
[mapped_data.get(field, False) for field in CHECKIN_FIELDS]
):
document = self.env["res.partner.id_number"].search(
[
("name", "=", mapped_data["document_number"]),
],
limit=1,
)
if document:
mapped_data = self._complete_mapped_from_partner(document, mapped_data)
return mapped_data
def _calc_expedition_date(
@@ -220,23 +260,54 @@ class PmsProperty(models.Model):
result = date_of_expiry - relativedelta(years=10)
return result if result else False
def _get_document_type(self, klippa_type, country_id):
document_type_id = False
document_type_ids = self.env["res.partner.id_category"].search(
[
("klippa_code", "=", klippa_type),
]
)
if not document_type_ids:
raise ValidationError(_(f"Document type not found: {klippa_type}"))
def _get_number_keys(self, document_data):
# Heuristic to identify the mapping of document_number and document_support_number
# with respect to the personal_number and document_number fields of klippa
# If the klippa document type is "I", and it is Spanish, then the personal_number
# we map it against document_number and document_number against document_support_number
# otherwise, the document_number we map against document_number and the personal_number
# against document_support_number
key_document_number = "document_number"
key_personal_number = "document_support_number"
if (
document_data.get("document_type", False)
and document_data.get("document_type").get("value") == "I"
and document_data.get("issuing_country", False)
and document_data.get("issuing_country").get("value") == "ESP"
):
key_document_number = "document_support_number"
key_personal_number = "document_number"
return (key_document_number, key_personal_number)
if len(document_type_ids) > 1:
document_type_id = document_type_ids.filtered(
lambda r: country_id in r.country_ids.ids
def _get_document_type(self, klippa_type, country_id):
# If we hace the issuing country, and document type is configured in the system
# to be used with the country, we use the country to get the document type
# If have issuing country and not found document type, we search a document type
# without country
# If not have issuing country, we search the document only by klippa code
document_type = False
domain = [("klippa_code", "=", klippa_type)]
if country_id:
domain.append(("country_ids", "in", country_id))
document_type = self.env["res.partner.id_category"].search(domain, limit=1)
if not document_type and country_id:
document_type = self.env["res.partner.id_category"].search(
[
("klippa_code", "=", klippa_type),
("country_ids", "=", False),
],
limit=1,
)
if not document_type_id:
document_type_id = document_type_ids[0]
return document_type_id[0]
elif not document_type:
document_type = self.env["res.partner.id_category"].search(
[
("klippa_code", "=", klippa_type),
],
limit=1,
)
if not document_type:
document_type = self.env.ref("pms.document_type_identification_document")
return document_type[0] if document_type else False
def _get_country_id(self, country_code):
if not country_code:
@@ -346,86 +417,103 @@ class PmsProperty(models.Model):
)
if address_data_dict.get("residence_city"):
params["city"] = address_data_dict["residence_city"]
# Try to complete the address with Nominatim API
try:
params = self._get_nominatim_address(params, street_name, mapped_data)
except Exception as e:
_logger.error(e)
mapped_data["nominatim_status"] = "error"
mapped_data["nominatim_response"] = str(e)
return mapped_data
def _get_nominatim_address(self, params, street_name, mapped_data):
if street_name:
# Clean street name with mains words
street_words = street_name.split(" ")
params["street"] = " ".join(
[word for word in street_words if len(word) > 2]
)
location = requests.get(NOMINATIM_URL, params=params)
if not location.json() or location.status_code != 200:
# If not found address, pop the street to try again
if street_name:
# Clean street name with mains words
street_words = street_name.split(" ")
params["street"] = " ".join(
[word for word in street_words if len(word) > 2]
)
location = requests.get(NOMINATIM_URL, params=params)
if not location.json() or location.status_code != 200:
# If not found address, pop the street to try again
if street_name:
params.pop("street")
location = requests.get(NOMINATIM_URL, params=params)
if location.json() and location.status_code == 200:
location = location.json()[0]
_logger.info(location)
if not mapped_data.get("zip", False):
mapped_data["zip"] = location.get("address", {}).get(
"postcode", False
params.pop("street")
location = requests.get(NOMINATIM_URL, params=params)
if location.json() and location.status_code == 200:
mapped_data["nominatim_response"] = location.json()
mapped_data["nominatim_status"] = "success"
location = location.json()[0]
_logger.info(location)
if not mapped_data.get("zip", False):
mapped_data["zip"] = location.get("address", {}).get("postcode", False)
if mapped_data["zip"]:
zip_code = self.env["res.city.zip"].search(
[("name", "=", mapped_data["zip"])]
)
if mapped_data["zip"]:
zip_code = self.env["res.city.zip"].search(
[("name", "=", mapped_data["zip"])]
if zip_code:
mapped_data["residence_city"] = zip_code.city_id.name
mapped_data["country_state"] = zip_code.city_id.state_id.id
mapped_data[
"country_id"
] = zip_code.city_id.state_id.country_id.id
if not mapped_data.get("country_id", False):
country_record = self.env["res.country"].search(
[
(
"code",
"=",
location.get("address", {})
.get("country_code", False)
.upper(),
)
if zip_code:
mapped_data["residence_city"] = zip_code.city_id.name
mapped_data["country_state"] = zip_code.city_id.state_id.id
mapped_data[
"country_id"
] = zip_code.city_id.state_id.country_id.id
if not mapped_data.get("country_id", False):
country_record = self.env["res.country"].search(
[
(
"code",
"=",
location.get("address", {})
.get("country_code", False)
.upper(),
)
]
]
)
if not country_record and location.get("address", {}).get(
"country", False
):
country_match = process.extractOne(
location.get("address", {}).get("country", False),
self.env["res.country"]
.with_context(lang="en_US")
.search([])
.mapped("name"),
)
if not country_record and location.get("address", {}).get(
"country", False
):
country_match = process.extractOne(
location.get("address", {}).get("country", False),
if country_match[1] >= 90:
country_record = (
self.env["res.country"]
.with_context(lang="en_US")
.search([])
.mapped("name"),
.search([("name", "=", country_match[0])])
)
if country_match[1] >= 90:
country_record = (
self.env["res.country"]
.with_context(lang="en_US")
.search([("name", "=", country_match[0])])
)
mapped_data["country_id"] = country_record.id
if not mapped_data.get("country_state", False):
state_name = (
location.get("address", {}).get("province")
if location.get("address", {}).get("province")
else location.get("address", {}).get("state")
mapped_data["country_id"] = country_record.id
if not mapped_data.get("country_state", False):
state_name = (
location.get("address", {}).get("province")
if location.get("address", {}).get("province")
else location.get("address", {}).get("state")
)
if state_name:
country_state_record = process.extractOne(
state_name,
self.env["res.country.state"].search([]).mapped("name"),
)
if state_name:
country_state_record = process.extractOne(
state_name,
self.env["res.country.state"].search([]).mapped("name"),
if country_state_record[1] >= 90:
country_state = self.env["res.country.state"].search(
[("name", "=", country_state_record[0])]
)
if country_state_record[1] >= 90:
country_state = self.env["res.country.state"].search(
[("name", "=", country_state_record[0])]
)
mapped_data["country_state"] = country_state.id
if not mapped_data.get("residence_city", False):
mapped_data["residence_city"] = location.get("address", {}).get(
"city", False
)
if not mapped_data.get("residence_street", False):
mapped_data["residence_street"] = location.get("address", {}).get(
"road", False
)
mapped_data["country_state"] = country_state.id
if not mapped_data.get("residence_city", False):
mapped_data["residence_city"] = location.get("address", {}).get(
"city", False
)
if not mapped_data.get("residence_street", False):
mapped_data["residence_street"] = location.get("address", {}).get(
"road", False
)
return mapped_data
def _complete_mapped_from_partner(self, document, mapped_data):
for key, field in CHECKIN_FIELDS.items():
if not mapped_data.get(key, False) and document.mapped(field)[0]:
mapped_data[key] = document.mapped(field)[0]
return mapped_data

8
pms_ocr_klippa/views/klippa_log_views.xml
View File

@@ -27,6 +27,10 @@
<field name="final_status" />
<field name="request_id" />
</group>
<group>
<field name="request_duration" widget="float_time" />
<field name="mapped_duration" widget="float_time" />
</group>
<notebook>
<page string="Klippa Response">
<field name="klippa_response" />
@@ -34,6 +38,10 @@
<page string="Service Response">
<field name="service_response" />
</page>
<page string="Nominatim Response">
<field name="nominatim_status" />
<field name="nominatim_response" />
</page>
<page string="Front Imgage">
<field name="image_base64_front" />
</page>