Homec4science
Diffusion cop-mining-participants (master) Browse

cop-mining-participants/code/scripts/generate_complete_dataset.py
f94770541d4bmaster

Last Change
Raw File

generate_complete_dataset.py
View Options

import pandas as pd
import country_converter as coco
import editdistance
import os
import re
import partlistproc.MeetingAnalyzer as Ana
import partlistproc.MeetingAnalyzerFactory as AnaFac
""" This script generates one csv file containing all the participants of all
the available meetings (information taken from metadata file)
"""
def is_male(name):
return any(title in name for title in Ana.MeetingAnalyzer.masculine_salutory_addresses)
def is_female(name):
return any(title in name for title in Ana.MeetingAnalyzer.feminine_salutory_addresses)
def has_title(name):
return any(title in name for title in Ana.MeetingAnalyzer.titles)
def has_no_title(name):
return not has_title(name)
def get_role(description):
# EDIT: I had to redesign this to make keywords of several words possible (28.11.20)
description = str(description)
if description == "nan":
return "no description"
splitted = re.split('[; ]{1}', description)
for key_line in roles_dict.keys():
keywords = re.split(" ", key_line)
if(str(keywords[0]) in splitted or (str(keywords[0])).lower() in
splitted):
if(len(keywords) == 1):
return roles_dict[key_line]
else:
# keyword is more than one word: check the others
found_word = str(keywords[0])
if found_word not in splitted:
found_word = found_word.lower()
index = splitted.index(found_word)
size = len(splitted)
for i in range(1, len(keywords)):
if index + i >= size or str(keywords[i]).lower() != str(splitted[index + i]).lower():
continue
return roles_dict[key_line]
return "no keyword found"
def clear_name(name):
"""removes all salutory addresses and titles from a given name
Args:
name (str): the name to be cleared
"""
cleared_name = name
while cleared_name.startswith(Ana.MeetingAnalyzer.salutory_addresses):
startindex = cleared_name.find(" ")
if startindex == -1:
return cleared_name
startindex += 1
cleared_name = cleared_name[startindex:]
return cleared_name.lower()
short_country_names = dict()
short_country_names["european union"] = "European Union"
short_country_names["european community"] = "European Union"
def simplify_country_name(affiliation):
if affiliation in short_country_names:
return short_country_names[affiliation]
else:
# None for not found makes that it returns the input value
converted = (coco.convert(names=[affiliation], to="name_short",
not_found=None))
if isinstance(converted, list):
converted = converted[0]
short_country_names[affiliation] = converted
return converted
def is_fossil_fuel_associated(words):
"""checks if the given string contains a fossil fuel industry keyword
Args:
words (str): the string to be tested for keywords
"""
splitted = re.split('[; ]{1}', (str(words)).lower())
for keyword in fossil_fuel_keywords:
if keyword in splitted:
return True
return False
# pre-processing
# extract the list of roles
roles_dict = {}
roles_file = open("../data/dictionaries/role_keywords.txt", "r", encoding="utf-8")
role_lines = roles_file.readlines()
current_role = ""
for line in role_lines:
if "\n" in line:
line = line[:line.index("\n")]
if line.startswith("["):
if not line.endswith("]"):
raise KeyError("Format on line {} was incorrect".format(line))
current_role = line[1:len(line) - 1]
else:
if line != "":
roles_dict[line] = current_role
# extract the list of fossil fuel industry keywords
fossil_fuel_keywords = []
ff_file = open("../data/dictionaries/fossil_fuel_industry_keywords.txt", "r",
encoding="utf-8")
ff_lines = ff_file.readlines()
for line in ff_lines:
if "\n" in line:
line = line[:line.index("\n")]
if line != "":
fossil_fuel_keywords.append(line.lower())
# begin with the real processing
complete_data = pd.DataFrame(columns={
"meeting",
"name",
"gender",
"has_title",
"affiliation",
"affiliation_category",
"role",
"fossil_fuel_industry",
"description"})
metadata = pd.read_csv("../data/meetings_metadata.csv")
for label in metadata["label"]:
datafile_name = "../results/participants-csv/participants_" + label + ".csv"
if label in AnaFac.MeetingAnalyzerFactory.french_meetings:
datafile_name = "../results/participants-csv/participants_" + label + "-en.csv"
if not os.path.isfile(datafile_name):
os.system("python extract_participants.py " + label)
os.system("python translate_list_fr_en.py " + label)
# generate the data if not yet available
if not os.path.isfile(datafile_name):
os.system("python extract_participants.py " + label)
# open the data from this cop
cop_data = pd.read_csv(datafile_name, encoding="utf-8-sig")
# add its data to the complete dataframe
cop_data["meeting"] = label
# determine gender
cop_data.loc[cop_data.name.apply(is_male), "gender"] = "m"
cop_data.loc[cop_data.name.apply(is_female), "gender"] = "f"
# determine title (if any)
cop_data.loc[cop_data.name.apply(has_title), "has_title"] = 1
cop_data.loc[cop_data.name.apply(has_no_title), "has_title"] = 0
# define the role
cop_data["role"] = cop_data["description"].apply(get_role)
# define the association to fossil fuel industry
cop_data["fossil_fuel_industry"] = 0
cop_data.loc[cop_data.description.apply(is_fossil_fuel_associated), "fossil_fuel_industry"] = 1
cop_data.loc[cop_data.affiliation.apply(is_fossil_fuel_associated), "fossil_fuel_industry"] = 1
# clear up the name
cop_data["name"] = cop_data["name"].apply(clear_name)
# unify the country names
cop_data.loc[cop_data.affiliation_category.apply(lambda p: p == "parties"), "affiliation"] = cop_data.loc[cop_data.affiliation_category.apply(lambda p: p == "parties"), "affiliation"].apply(simplify_country_name)
print(label)
print(cop_data[:5])
complete_data = complete_data.append(cop_data, ignore_index=True)
# only for a short time
grouped_by_role = complete_data.groupby("role")
for role, rest in grouped_by_role:
print(f"{role}: {len(rest)} participants found")
print(f"Country names map of length {len(short_country_names)}")
print(short_country_names)
short_country_names_cleaned = {k: v for (k, v) in short_country_names.items() if k != v}
country_set = set(short_country_names_cleaned.values())
print(f"Set of length {len(country_set)}")
print(country_set)
f = open("../data/dictionaries/valid_countries.txt", "w")
for country in country_set:
if country != "Vatican":
f.write(str(country) + "\n")
f.write("european union")
f.close()
# generate the output file
complete_data.to_csv("../results/complete_dataset.csv", encoding="utf-8-sig",
index=False)
c4science · Help