13789084875

Committed 11 Mar 2025 01:21PM UTC coverage: 38.661%. First build

Build # 13789084875

Build Type

Pull #170

github

Committed by

web-flow

Commit Message

Merge 72fa00dd1 into df9d78356

Pull Request Pull Request #170: Put the merged dashboard work live

Run Details

778 of 1567 new or added lines in 25 files covered. (49.65%)

878 of 2271 relevant lines covered (38.66%)

0.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

12.04

/statsrunner/loop.py

import inspect
import json
import os
import sys
import traceback

from lxml import etree

import statsrunner.aggregate
import statsrunner.shared
from statsrunner.common import decimal_default, sort_keys


def call_stats(this_stats, args):
    """Create dictionary of enabled stats for this_stats object.

    Args:
      this_stats (cls): stats_module that specifies calculations for relevant input, processed by internal methods of process_file().

      args: Object containing program run options (set by CLI arguments at runtime. See __init__ for more details).
    """
    this_out = {}
    # For each method within this_stats object check the method is an enabled stat, if it is not enabled continue to next method.
    for name, function in inspect.getmembers(this_stats, predicate=inspect.ismethod):
        # If method is an enabled stat, add the method to the this_out dictionary, unless the exception criteria are met.
        if not statsrunner.shared.use_stat(this_stats, name):
            continue
        try:
            this_out[name] = function()
        except KeyboardInterrupt:
            exit()
        except Exception:
            traceback.print_exc(file=sys.stdout)
    if args.debug:
        print(this_out)
    return this_out


def process_file(*args):
    """Create output file path or write output file."""
    args = args[0]
    inputfile = args[0]
    output_dir = args[1]
    folder = args[2]
    xmlfile = args[3]
    args = args[4]
    import importlib

    # Python module to import stats from defaults to stats.dashboard
    stats_module = importlib.import_module(args.stats_module)
    # When args.verbose_loop is true, create directory and set outputfile according to loop path.
    if args.verbose_loop:
        try:
            os.makedirs(os.path.join(output_dir, "loop", folder))
        except OSError:
            pass
        outputfile = os.path.join(output_dir, "loop", folder, xmlfile)
    # If args.verbose_loop is false, set outputfile according to aggregated-file path.
    else:
        outputfile = os.path.join(output_dir, "aggregated-file", folder, xmlfile)

    # If default args is set to only create new files, check for existing file and return early.
    if args.new:
        if os.path.exists(outputfile):
            return
    # If default args are not set to only create new files try setting file_size to size of file in bytes.
    try:
        file_size = os.stat(inputfile).st_size
        # If the file size is greater than the registry limit, set stats_json file value to 'too large'.
        # Registry limit: https://github.com/IATI/ckanext-iati/blob/6ec9109826aec42e4fc9297db198753f83a48f80/ckanext/iati/archiver.py#L34-L36
        if file_size > 60000000:
            stats_json = {"file": {"toolarge": 1, "file_size": file_size}, "elements": []}
        # If file size is within limit, set doc to the value of the complete inputfile document, and set root to the root of element tree for doc.
        else:
            doc = etree.parse(inputfile)
            root = doc.getroot()

            def process_stats_file(FileStats):
                """Set object elements and pass to call_stats()."""
                file_stats = FileStats()
                file_stats.doc = doc
                file_stats.root = root
                file_stats.strict = args.strict
                file_stats.context = "in " + inputfile
                file_stats.fname = os.path.basename(inputfile)
                file_stats.inputfile = inputfile
                return call_stats(file_stats, args)

            def process_stats_element(ElementStats, tagname=None):
                """Generate object elements and yield to call_stats()."""
                for element in root:
                    if tagname and tagname != element.tag:
                        continue
                    element_stats = ElementStats()
                    element_stats.element = element
                    element_stats.strict = args.strict
                    element_stats.context = "in " + inputfile
                    element_stats.today = args.today
                    yield call_stats(element_stats, args)

            def process_stats(FileStats, ElementStats, tagname=None):
                """Create dictionary with processed stats_module objects.

                Args:
                    FileStats (cls): stats_module that contains calculations for an organisation or activity XML file.
                    ElementStats (cls): stats_module that contains raw stats calculations for a single organisation or activity.
                    tagname: Label for type of stats_module.

                Returns:
                    Dictionary with values that are dictionaries of the enabled stats for the file and elements being processed.
                """
                file_out = process_stats_file(FileStats)
                out = process_stats_element(ElementStats, tagname)
                return {"file": file_out, "elements": out}

            if root.tag == "iati-activities":
                stats_json = process_stats(stats_module.ActivityFileStats, stats_module.ActivityStats, "iati-activity")
            elif root.tag == "iati-organisations":
                stats_json = process_stats(
                    stats_module.OrganisationFileStats, stats_module.OrganisationStats, "iati-organisation"
                )
            else:
                stats_json = {"file": {"nonstandardroots": 1}, "elements": []}

    # If there is a ParseError print statement, then set stats_json file value according to whether the file size is zero.
    except etree.ParseError:
        print("Could not parse file {0}".format(inputfile))
        if os.path.getsize(inputfile) == 0:
            # Assume empty files are download errors, not invalid XML
            stats_json = {"file": {"emptyfile": 1}, "elements": []}
        else:
            stats_json = {"file": {"invalidxml": 1}, "elements": []}

    # If args.verbose_loop is true, assign value of list of stats_json element keys to stats_json elements key and write to json file.
    if args.verbose_loop:
        with open(outputfile, "w") as outfp:
            stats_json["elements"] = list(stats_json["elements"])
            json.dump(sort_keys(stats_json), outfp, indent=2, default=decimal_default)
    # If args.verbose_loop is not true, create aggregated-file json and return the subtotal dictionary of statsrunner.aggregate.aggregate_file().
    else:
        statsrunner.aggregate.aggregate_file(
            stats_module, stats_json, os.path.join(output_dir, "aggregated-file", folder, xmlfile)
        )


def loop_folder(folder, args, data_dir, output_dir):
    """Given a folder, returns a list of XML files in folder."""
    if not os.path.isdir(os.path.join(data_dir, folder)) or folder == ".git":
        return []
    files = []
    for xmlfile in os.listdir(os.path.join(data_dir, folder)):
        try:
            files.append((os.path.join(data_dir, folder, xmlfile), output_dir, folder, xmlfile, args))
        except UnicodeDecodeError:
            traceback.print_exc(file=sys.stdout)
            continue
    return files


def loop(args):
    """Loops through all specified folders to convert data to JSON output.

    Args:
        args: Object containing program run options (set by CLI arguments at runtime. See __init__ for more details).
    """
    if args.folder:
        files = loop_folder(args.folder, args, data_dir=args.data, output_dir=args.output)
    else:
        files = []
        for folder in os.listdir(args.data):
            files += loop_folder(folder, args, data_dir=args.data, output_dir=args.output)

    if args.multi > 1:
        from multiprocessing import Pool

        pool = Pool(args.multi)
        pool.map(process_file, files)
    else:
        list(map(process_file, files))

1	import inspect	1✔
2	import json	1✔
3	import os	1✔
4	import sys	1✔
5	import traceback	1✔
6
7	from lxml import etree	1✔
8
9	import statsrunner.aggregate	1✔
10	import statsrunner.shared	1✔
11	from statsrunner.common import decimal_default, sort_keys	1✔
12
13
14	def call_stats(this_stats, args):	1✔
15	"""Create dictionary of enabled stats for this_stats object.
16
17	Args:
18	this_stats (cls): stats_module that specifies calculations for relevant input, processed by internal methods of process_file().
19
20	args: Object containing program run options (set by CLI arguments at runtime. See __init__ for more details).
21	"""
22	this_out = {}	×
23	# For each method within this_stats object check the method is an enabled stat, if it is not enabled continue to next method.
24	for name, function in inspect.getmembers(this_stats, predicate=inspect.ismethod):	×
25	# If method is an enabled stat, add the method to the this_out dictionary, unless the exception criteria are met.
26	if not statsrunner.shared.use_stat(this_stats, name):	×
27	continue	×
28	try:	×
29	this_out[name] = function()	×
30	except KeyboardInterrupt:	×
31	exit()	×
NEW 32	except Exception:	×
33	traceback.print_exc(file=sys.stdout)	×
34	if args.debug:	×
NEW 35	print(this_out)	×
36	return this_out	×
37
38
39	def process_file(*args):	1✔
40	"""Create output file path or write output file."""
NEW 41	args = args[0]	×
NEW 42	inputfile = args[0]	×
NEW 43	output_dir = args[1]	×
NEW 44	folder = args[2]	×
NEW 45	xmlfile = args[3]	×
NEW 46	args = args[4]	×
47	import importlib	×
48
49	# Python module to import stats from defaults to stats.dashboard
50	stats_module = importlib.import_module(args.stats_module)	×
51	# When args.verbose_loop is true, create directory and set outputfile according to loop path.
52	if args.verbose_loop:	×
53	try:	×
NEW 54	os.makedirs(os.path.join(output_dir, "loop", folder))	×
55	except OSError:	×
56	pass	×
NEW 57	outputfile = os.path.join(output_dir, "loop", folder, xmlfile)	×
58	# If args.verbose_loop is false, set outputfile according to aggregated-file path.
59	else:
NEW 60	outputfile = os.path.join(output_dir, "aggregated-file", folder, xmlfile)	×
61
62	# If default args is set to only create new files, check for existing file and return early.
63	if args.new:	×
64	if os.path.exists(outputfile):	×
65	return	×
66	# If default args are not set to only create new files try setting file_size to size of file in bytes.
67	try:	×
68	file_size = os.stat(inputfile).st_size	×
69	# If the file size is greater than the registry limit, set stats_json file value to 'too large'.
70	# Registry limit: https://github.com/IATI/ckanext-iati/blob/6ec9109826aec42e4fc9297db198753f83a48f80/ckanext/iati/archiver.py#L34-L36
71	if file_size > 60000000:	×
NEW 72	stats_json = {"file": {"toolarge": 1, "file_size": file_size}, "elements": []}	×
73	# If file size is within limit, set doc to the value of the complete inputfile document, and set root to the root of element tree for doc.
74	else:
75	doc = etree.parse(inputfile)	×
76	root = doc.getroot()	×
77
78	def process_stats_file(FileStats):	×
79	"""Set object elements and pass to call_stats()."""
80	file_stats = FileStats()	×
81	file_stats.doc = doc	×
82	file_stats.root = root	×
83	file_stats.strict = args.strict	×
NEW 84	file_stats.context = "in " + inputfile	×
85	file_stats.fname = os.path.basename(inputfile)	×
86	file_stats.inputfile = inputfile	×
87	return call_stats(file_stats, args)	×
88
89	def process_stats_element(ElementStats, tagname=None):	×
90	"""Generate object elements and yield to call_stats()."""
91	for element in root:	×
92	if tagname and tagname != element.tag:	×
93	continue	×
94	element_stats = ElementStats()	×
95	element_stats.element = element	×
96	element_stats.strict = args.strict	×
NEW 97	element_stats.context = "in " + inputfile	×
98	element_stats.today = args.today	×
99	yield call_stats(element_stats, args)	×
100
101	def process_stats(FileStats, ElementStats, tagname=None):	×
102	"""Create dictionary with processed stats_module objects.
103
104	Args:
105	FileStats (cls): stats_module that contains calculations for an organisation or activity XML file.
106	ElementStats (cls): stats_module that contains raw stats calculations for a single organisation or activity.
107	tagname: Label for type of stats_module.
108
109	Returns:
110	Dictionary with values that are dictionaries of the enabled stats for the file and elements being processed.
111	"""
112	file_out = process_stats_file(FileStats)	×
113	out = process_stats_element(ElementStats, tagname)	×
NEW 114	return {"file": file_out, "elements": out}	×
115
NEW 116	if root.tag == "iati-activities":	×
NEW 117	stats_json = process_stats(stats_module.ActivityFileStats, stats_module.ActivityStats, "iati-activity")	×
NEW 118	elif root.tag == "iati-organisations":	×
NEW 119	stats_json = process_stats(	×
120	stats_module.OrganisationFileStats, stats_module.OrganisationStats, "iati-organisation"
121	)
122	else:
NEW 123	stats_json = {"file": {"nonstandardroots": 1}, "elements": []}	×
124
125	# If there is a ParseError print statement, then set stats_json file value according to whether the file size is zero.
126	except etree.ParseError:	×
NEW 127	print("Could not parse file {0}".format(inputfile))	×
128	if os.path.getsize(inputfile) == 0:	×
129	# Assume empty files are download errors, not invalid XML
NEW 130	stats_json = {"file": {"emptyfile": 1}, "elements": []}	×
131	else:
NEW 132	stats_json = {"file": {"invalidxml": 1}, "elements": []}	×
133
134	# If args.verbose_loop is true, assign value of list of stats_json element keys to stats_json elements key and write to json file.
135	if args.verbose_loop:	×
NEW 136	with open(outputfile, "w") as outfp:	×
NEW 137	stats_json["elements"] = list(stats_json["elements"])	×
NEW 138	json.dump(sort_keys(stats_json), outfp, indent=2, default=decimal_default)	×
139	# If args.verbose_loop is not true, create aggregated-file json and return the subtotal dictionary of statsrunner.aggregate.aggregate_file().
140	else:
NEW 141	statsrunner.aggregate.aggregate_file(	×
142	stats_module, stats_json, os.path.join(output_dir, "aggregated-file", folder, xmlfile)
143	)
144
145
146	def loop_folder(folder, args, data_dir, output_dir):	1✔
147	"""Given a folder, returns a list of XML files in folder."""
NEW 148	if not os.path.isdir(os.path.join(data_dir, folder)) or folder == ".git":	×
149	return []	×
150	files = []	×
151	for xmlfile in os.listdir(os.path.join(data_dir, folder)):	×
152	try:	×
NEW 153	files.append((os.path.join(data_dir, folder, xmlfile), output_dir, folder, xmlfile, args))	×
154	except UnicodeDecodeError:	×
155	traceback.print_exc(file=sys.stdout)	×
156	continue	×
157	return files	×
158
159
160	def loop(args):	1✔
161	"""Loops through all specified folders to convert data to JSON output.
162
163	Args:
164	args: Object containing program run options (set by CLI arguments at runtime. See __init__ for more details).
165	"""
166	if args.folder:	×
167	files = loop_folder(args.folder, args, data_dir=args.data, output_dir=args.output)	×
168	else:
169	files = []	×
170	for folder in os.listdir(args.data):	×
171	files += loop_folder(folder, args, data_dir=args.data, output_dir=args.output)	×
172
173	if args.multi > 1:	×
174	from multiprocessing import Pool	×
175
176	pool = Pool(args.multi)	×
177	pool.map(process_file, files)	×
178	else:
NEW 179	list(map(process_file, files))	×

IATI / IATI-Stats / 13789084875

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous