from pathlib import Path import zipfile ''' to use the call_reports.py utility, download the call report zip files from https://cdr.ffiec.gov/public/PWS/DownloadBulkData.aspx and place in "./data/call-reports/" in the current working directory ''' def table(text): lines = text.split('\n') columns = [x for x in lines[0].split('\t') if x.strip()!= ''] data = lines[1:] results = [] for line in [x for x in data if x.strip() != '']: values = line.split('\t') item = {} results.append(item) for index, column in enumerate(columns): item[column] = values[index] pass return results def details(file): if file == 'Readme.txt': pass if file.lower() == 'readme.txt': return {"name":file,"date":None, "qualifier":None} split1 = file.split('.') temp1 = split1[0] start_index = temp1.find('(') qual = None if start_index != -1: # Slice from that index to the very end qual = temp1[start_index:] temp2 = temp1.split('(', 1)[0] split2 = temp2.split(' ') date = split2.pop() year = date[-4:] day = date[2:4] month= date[0:2] name = split2.pop() result = {"name":name, "date":year+'-'+month+'-'+day, "qualifier":qual, "file":file} return result def query(process,filter=lambda x:True, read=True): target_dir = Path("./data/call-reports") # Check if the directory exists and list files if target_dir.exists() and target_dir.is_dir(): files = [f.name for f in target_dir.iterdir() if f.is_file()] for file in files: with zipfile.ZipFile('./data/call-reports/'+file, 'r') as zf: file_list = [x for x in zf.namelist() if filter(x)] for file in file_list: if file.lower() != 'readme.txt': file_content = None if read==True: # Get all contents as a bytes string file_bytes = zf.read(file) # Convert bytes to string file_content = file_bytes.decode('utf-8') process(file_content, file) pass pass pass pass pass pass map = None def id_map(): global map if map != None: return map map = {} def filter(file): name = details(file)['name'] if name == 'POR': return True return False def process1(text, file): tab = table(text) for item in tab: map[item['"IDRSSD"']] = item pass pass query(process1, filter) return map def files(): results = set() target_dir = Path("./data/call-reports") # Check if the directory exists and list files if target_dir.exists() and target_dir.is_dir(): files = [f.name for f in target_dir.iterdir() if f.is_file()] for file in files: with zipfile.ZipFile('./data/call-reports/'+file, 'r') as zf: file_list = [details(x)['name'] for x in zf.namelist()] for fl in [x for x in file_list if x.lower()!='readme.txt']: results.add(fl) pass pass return list(results) def fields(file): results = set() def process(text, file): nonlocal results lines = text.split('\n') line = lines[0] split = line.split('\t') for item in split: results.add(item) pass def filter(file1): dets = details(file1) if dets['name'] == file: return True return False query(process, filter) return list(results) def banks(): map = id_map() results = [] for key in map: results.append(map[key]) return results def get(id, fields): if not isinstance(fields, list): fields = [fields] map = {} splits = [field.split('.') for field in fields] files = [split[0] for split in splits] def process(text, file): date = details(file)['date'] tab = table(text) for item in tab: if item['"IDRSSD"'] == id: record = {"date":date, "id":id} for split in splits: if split[0] == details(file)['name']: field = split[1] record[field] = item[split[1]] pass pass map[date] = record pass pass def filter(file): nonlocal files return details(file)['name'] in files query(process, filter) results = [] for date in map: results.append(map[date]) return results if __name__ == '__main__': fls = files() test_fields = fields('RCM') def test(text, file): det =details(file) tab = table(text) pass bks = banks() print(bks) test2 = get('1351','RCM.RCONN523') pass