Skip to content

Commit

Permalink
statistics.py: Count interested party types
Browse files Browse the repository at this point in the history
  • Loading branch information
Ed (ODSC) committed Nov 14, 2024
1 parent a90e648 commit bc8ebdf
Show file tree
Hide file tree
Showing 4 changed files with 86 additions and 55 deletions.
100 changes: 50 additions & 50 deletions libcovebods/tasks/checks.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -863,14 +863,14 @@ def does_apply_to_schema(lib_cove_bods_config, schema_object) -> bool:
def check_statement_first_pass(self, statement):
if "annotations" in statement and isinstance(statement["annotations"], list):
for annotation in statement["annotations"]:
print(annotation)
#print(annotation)
if (
isinstance(annotation, dict)
and "creationDate" in annotation
and annotation["creationDate"]
):
creation_date = parse_date_field(annotation["creationDate"])
print(creation_date)
#print(creation_date)
if creation_date and creation_date > datetime.now().date():
self._additional_check_results.append(
{
Expand Down Expand Up @@ -927,7 +927,7 @@ def check_person_statement_first_pass(self, statement):
and "deathDate" in statement["recordDetails"]
and statement["recordDetails"]["deathDate"]
):
print("Death date:", statement["recordDetails"]["deathDate"])
#print("Death date:", statement["recordDetails"]["deathDate"])
death_date = parse_date_field(statement["recordDetails"]["deathDate"])
if death_date:
if (
Expand Down Expand Up @@ -1111,16 +1111,16 @@ def does_apply_to_schema(lib_cove_bods_config, schema_object) -> bool:
return schema_object.is_schema_version_equal_to_or_greater_than("0.4")

def check_ownership_or_control_statement_first_pass(self, statement):
print("Checking Share Values:")
#print("Checking Share Values:")
if (
"recordDetails" in statement
and isinstance(statement["recordDetails"], dict)
and "interests" in statement["recordDetails"]
and isinstance(statement["recordDetails"]["interests"], list)
):
print("Here!")
#print("Here!")
for interest in statement["recordDetails"]["interests"]:
print(interest)
#print(interest)
if "share" in interest and isinstance(interest["share"], dict):
share = interest["share"]
if "exclusiveMinimum" in share and "minimum" in share:
Expand All @@ -1142,7 +1142,7 @@ def check_ownership_or_control_statement_first_pass(self, statement):
}
)
if "exact" in share and share["exact"]:
print("Here!")
#print("Here!")
if any(
[
limit in share
Expand Down Expand Up @@ -1186,14 +1186,14 @@ def check_ownership_or_control_statement_first_pass(self, statement):
if "maximum" in share
else float(share["exclusiveMaximum"])
)
print(
"Min:",
min_val,
"Max:",
max_val,
"Comp:",
max_val >= min_val,
)
#print(
# "Min:",
# min_val,
# "Max:",
# max_val,
# "Comp:",
# max_val >= min_val,
#)
if not max_val >= min_val:
self._additional_check_results.append(
{
Expand Down Expand Up @@ -1337,7 +1337,7 @@ def check_statement_first_pass(self, statement):
self._statements[statement["statementId"]] += 1

def check_statement_second_pass(self, statement):
print(self._statements)
#print(self._statements)
if "statementId" in statement:
if self._statements[statement["statementId"]] > 1:
self._statements[statement["statementId"]] -= 1
Expand Down Expand Up @@ -1549,15 +1549,15 @@ def check_ownership_or_control_statement_second_pass(self, statement):
]
== "person"
):
print(
"Check:",
self._records[
statement["recordDetails"][
"interestedParty"
]
],
"person",
)
#print(
# "Check:",
# self._records[
# statement["recordDetails"][
# "interestedParty"
# ]
# ],
# "person",
#)
self._additional_check_results.append(
{
"type": "interest_beneficial_ownership_interested_party_not_person",
Expand Down Expand Up @@ -1620,29 +1620,29 @@ def check_statement_first_pass(self, statement):
self._records[statement["recordId"]] = (record_type, record_type_type)

def check_ownership_or_control_statement_second_pass(self, statement):
print("Here 1")
#print("Here 1")
if "recordDetails" in statement and isinstance(
statement["recordDetails"], dict
):
print("Here 2")
#print("Here 2")
if "interests" in statement["recordDetails"] and isinstance(
statement["recordDetails"]["interests"], list
):
print("Here 3")
#print("Here 3")
for interest in statement["recordDetails"]["interests"]:
print("Interest:", interest)
#print("Interest:", interest)
if "type" in interest and interest["type"] in (
"nominee",
"nominator",
):
print("Here 4")
#print("Here 4")
if "subject" in statement["recordDetails"] and isinstance(
statement["recordDetails"]["subject"], str
):
print(
"Here 5",
self._records[statement["recordDetails"]["subject"]],
)
#print(
# "Here 5",
# self._records[statement["recordDetails"]["subject"]],
#)
if (
not self._records[
statement["recordDetails"]["subject"]
Expand All @@ -1662,7 +1662,7 @@ def check_ownership_or_control_statement_second_pass(self, statement):
entity_type = self._records[
statement["recordDetails"]["subject"]
][1]
print("Entity type:", entity_type)
#print("Entity type:", entity_type)
if (
not entity_type
or not isinstance(statement["recordDetails"], dict)
Expand All @@ -1685,14 +1685,14 @@ def check_ownership_or_control_statement_second_pass(self, statement):
"trustee",
"protector",
):
print("Here 4")
#print("Here 4")
if "subject" in statement["recordDetails"] and isinstance(
statement["recordDetails"]["subject"], str
):
print(
"Here 5",
self._records[statement["recordDetails"]["subject"]],
)
#print(
# "Here 5",
# self._records[statement["recordDetails"]["subject"]],
#)
if (
not self._records[
statement["recordDetails"]["subject"]
Expand All @@ -1712,7 +1712,7 @@ def check_ownership_or_control_statement_second_pass(self, statement):
entity_type = self._records[
statement["recordDetails"]["subject"]
][1]
print("Entity type:", entity_type)
#print("Entity type:", entity_type)
if (
not entity_type
or not isinstance(statement["recordDetails"], dict)
Expand Down Expand Up @@ -1782,14 +1782,14 @@ def check_ownership_or_control_statement_second_pass(self, statement):
self._records[statement["recordDetails"]["interestedParty"]]
> self._records[statement["recordId"]]
):
print(
"Ordering:",
self._records[
statement["recordDetails"]["interestedParty"]
],
self._records[statement["recordId"]],
self._records,
)
#print(
# "Ordering:",
# self._records[
# statement["recordDetails"]["interestedParty"]
# ],
# self._records[statement["recordId"]],
# self._records,
#)
self._additional_check_results.append(
{
"type": "relationship_interested_party_not_before_relationship_in_dataset",
Expand Down Expand Up @@ -1819,7 +1819,7 @@ def check_person_statement_first_pass(self, statement):
and isinstance(statement["recordDetails"]["identifiers"], list)
):
for identifier in statement["recordDetails"]["identifiers"]:
print("Got here!")
#print("Got here!")
if not (
"scheme" in identifier
and identifier["scheme"].count("-") == 1
Expand Down
29 changes: 29 additions & 0 deletions libcovebods/tasks/statistics.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,12 @@ def does_apply_to_schema(lib_cove_bods_config, schema_object) -> bool:

def __init__(self, lib_cove_bods_config, schema_object):
super().__init__(lib_cove_bods_config, schema_object)
self.entity_record_ids = []
self.person_record_ids = []
self.count_ownership_or_control_statement = 0
self.count_ownership_or_control_statement_interested_party_with_person = 0
self.count_ownership_or_control_statement_interested_party_with_entity = 0
self.count_ownership_or_control_statement_interested_party_with_unspecified = 0
self.count_ownership_or_control_statement_interested_party = 0
self.count_ownership_or_control_statement_interest_statement_types = {}
for (
Expand All @@ -332,6 +337,14 @@ def __init__(self, lib_cove_bods_config, schema_object):
defaultdict(int)
)

def check_entity_statement_first_pass(self, statement):
if "recordId" in statement:
self.entity_record_ids.append(statement["recordId"])

def check_person_statement_first_pass(self, statement):
if "recordId" in statement:
self.person_record_ids.append(statement["recordId"])

def check_ownership_or_control_statement_first_pass(self, statement):
try:
year = int(statement.get("statementDate", "").split("-")[0])
Expand Down Expand Up @@ -370,9 +383,25 @@ def check_ownership_or_control_statement_first_pass(self, statement):
if "statementDate" in statement:
self.count_ownership_or_control_statement_by_year[year] += 1

def check_ownership_or_control_statement_second_pass(self, statement):
if "recordDetails" in statement and isinstance(
statement["recordDetails"], dict
):
interested_party = statement["recordDetails"].get("interestedParty")
if interested_party:
if interested_party in self.entity_record_ids:
self.count_ownership_or_control_statement_interested_party_with_entity += 1
if interested_party in self.person_record_ids:
self.count_ownership_or_control_statement_interested_party_with_person += 1
if isinstance(interested_party, dict):
self.count_ownership_or_control_statement_interested_party_with_unspecified += 1

def get_statistics(self):
data = {
"count_ownership_or_control_statement": self.count_ownership_or_control_statement,
"count_ownership_or_control_statement_interested_party_with_person": self.count_ownership_or_control_statement_interested_party_with_person,
"count_ownership_or_control_statement_interested_party_with_entity": self.count_ownership_or_control_statement_interested_party_with_entity,
"count_ownership_or_control_statement_interested_party_with_unspecified": self.count_ownership_or_control_statement_interested_party_with_unspecified,
"count_ownership_or_control_statement_interested_party": self.count_ownership_or_control_statement_interested_party,
"count_ownership_or_control_statement_interest_statement_types": self.count_ownership_or_control_statement_interest_statement_types,
"count_ownership_or_control_statement_by_year": self.count_ownership_or_control_statement_by_year,
Expand Down
7 changes: 3 additions & 4 deletions tests/test_additional_fields_0_4_0.py
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
import os
import tempfile

from tests.api import bods_json_output

import libcovebods.data_reader
from libcovebods.schema import SchemaBODS
from libcovebods.config import LibCoveBODSConfig
from libcovebods.additionalfields import AdditionalFields
from libcovebods.config import LibCoveBODSConfig
from libcovebods.schema import SchemaBODS
from tests.api import bods_json_output


def test_additional_fields_1():
Expand Down
5 changes: 4 additions & 1 deletion tests/test_stat_counts_0_4.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ def test_sample_mode_valid_1():

results = bods_json_output(cove_temp_folder, json_filename, sample_mode=True)

print(results)
#print(results)

assert results["schema_version"] == "0.4"

Expand All @@ -29,6 +29,9 @@ def test_sample_mode_valid_1():
assert results['statistics']['count_person_statements'] == 50
assert results['statistics']['count_person_statements_types']['knownPerson'] == 50
assert results['statistics']['count_ownership_or_control_statement'] == 50
assert results['statistics']["count_ownership_or_control_statement_interested_party_with_person"] == 50
assert results['statistics']["count_ownership_or_control_statement_interested_party_with_entity"] == 0
assert results['statistics']["count_ownership_or_control_statement_interested_party_with_unspecified"] == 0
assert results['statistics']['count_ownership_or_control_statement_interested_party'] == 50
assert results['statistics']['count_ownership_or_control_statement_interest_statement_types']['shareholding'] == 50
assert results['statistics']['count_ownership_or_control_statement_by_year'][2020] == 50
Expand Down

0 comments on commit bc8ebdf

Please sign in to comment.