Merged in feature/adhoc-ops-scripts (pull request #275)
DRAFT: Feature/adhoc ops scripts * added adhoc ops scripts for pre-doczy work * Added first draft of cost automation script * Fixed dir name * Added Scheduler lambda * Added TX adhoc script for rerun + Updated cost automation script * Added some misc scripts * Merged main into feature/adhoc-ops-scripts * Aryan Ad Hoc Scripts Pushed * De Duplication Script added * Merged main into feature/adhoc-ops-scripts Approved-by: Umang Shailesh Mistry
This commit is contained in:
Michael McGuinness
committed by
Umang Shailesh Mistry
Umang Shailesh Mistry
parent
e8d231b801
commit
6c028ef892
@@ -0,0 +1,140 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"The first part of the process is to take a list of file names and their respective page counts and perform a simple de-duplication process that takes care of renaming files which are not actually duplicates and discarding those which are \"true\" duplicates. \n",
|
||||
"* We decide this based on the page counts. If the same file names have different page counts, they are not actually duplicates. However if the file name is same and the page count is same - they are classified as \"true\" duplicates.\n",
|
||||
"\n",
|
||||
"The Input CSV should essentially have two important columns - All the File Names, File Paths, and their Page Counts.\n",
|
||||
"\n",
|
||||
"Our output for duplicate file names should rename them as (1), (2), .... and so on.\n",
|
||||
"\n",
|
||||
"Follow the comments to understand the step by step procedure taken to handle de-duplication and conduct renaming."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# Importing pandas to deal with the csv as a dataframe\n",
|
||||
"import pandas as pd\n",
|
||||
"# Importing concurrent futures to support multithreading\n",
|
||||
"import concurrent.futures\n",
|
||||
"\n",
|
||||
"# Helper Function that takes in a group of files with the exact same file name and performs the core logic\n",
|
||||
"def process_file_group(group):\n",
|
||||
" # This first drops rows with same file name and same page count and then sorts the page counts in order\n",
|
||||
" group = group.drop_duplicates(subset=['File Name', 'Page Count']).sort_values('Page Count').reset_index(drop=True)\n",
|
||||
" # This is just a simple function to add the index as (n) in the correct place of the file name\n",
|
||||
" def add_suffix(file_name, idx):\n",
|
||||
" if file_name.endswith('.Pdf'):\n",
|
||||
" return file_name[:-4] + f'({idx + 1}).Pdf'\n",
|
||||
" elif file_name.endswith('.pdf'):\n",
|
||||
" return file_name[:-4] + f'({idx + 1}).pdf'\n",
|
||||
" else:\n",
|
||||
" return file_name + f'({idx + 1})'\n",
|
||||
" # This run the above function on all of the file names in the group and populates those in a new column\n",
|
||||
" group['New File Name'] = [add_suffix(file_name, idx) for idx, file_name in enumerate(group['File Name'])]\n",
|
||||
" return group\n",
|
||||
"\n",
|
||||
"input_csv = 'icertis/icertis_files.csv'\n",
|
||||
"df = pd.read_csv(input_csv)\n",
|
||||
"# This is a good trick to identify rows which have duplicate file names and keep only those in the dataframe\n",
|
||||
"duplicates_only = df[df.duplicated(subset=['File Name'], keep=False)]\n",
|
||||
"# The group by function is used to bring all the same file names in the same group so that we can process and relabel those\n",
|
||||
"grouped = duplicates_only.groupby(['File Name'])\n",
|
||||
"print(grouped.head())\n",
|
||||
"# This is a typical multithreading technique to run the process function on every group and store the results\n",
|
||||
"with concurrent.futures.ThreadPoolExecutor() as executor:\n",
|
||||
" results = executor.map(process_file_group, [group for _, group in grouped])\n",
|
||||
"\n",
|
||||
"# Now we simply put back the results into an output csv\n",
|
||||
"optimized_df = pd.concat(results)\n",
|
||||
"optimized_df.to_csv('icertis/icertis_duplicate_files.csv', index=False)\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"This the second step where we do two things - \n",
|
||||
"* figure out which files we are concerned about in the current processing and get a csv of those original file names.\n",
|
||||
"* Apply a custom 5-pager logic that does the following \n",
|
||||
" * For a specific file name, if we have a file that has greater than 5 additional pages compared to all other duplicates of that file, we only take that max page count file\n",
|
||||
" * If there are files within the 5 page range of the max page count file, we are not sure which file to consider and hence we add all those files\n",
|
||||
"\n",
|
||||
"We need to inputs here, one is the list of file names to consider and second is the previously generated csv of all duplicate files handled.\n",
|
||||
"\n",
|
||||
"Follow the comments to understand the step by step procedure taken"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# Importing pandas to deal with the csv as a dataframe\n",
|
||||
"import pandas as pd\n",
|
||||
"\n",
|
||||
"# Here we load the two input csvs and set up the output dataframe\n",
|
||||
"df_dup = pd.read_csv('duplicate_files_list.csv')\n",
|
||||
"# df_dup['File Name'] = df_dup['File Name'][:-4]\n",
|
||||
"df_priority = pd.read_csv('batch6_16_comparison_output.csv')\n",
|
||||
"df_priority = df_priority[df_priority['Only in Master Tracker'].str.contains(r'\\(\\d+\\)', na=False)]\n",
|
||||
"# df_priority['File Name'] = df_priority['File Name'][:-4]\n",
|
||||
"output_df = pd.DataFrame(columns=df_dup.columns)\n",
|
||||
"\n",
|
||||
"# Here we iterate over all the file names in the concerned file name list\n",
|
||||
"print(len(df_priority['Only in Master Tracker']))\n",
|
||||
"for file in df_priority['Only in Master Tracker']:\n",
|
||||
" # We extract the relevant rows from the duplicate files that match the name\n",
|
||||
" relevant_rows = df_dup[df_dup['New File Name'] == (file+\".Pdf\")]\n",
|
||||
" # If there is only one matching row, we have no logic needed and can just use that\n",
|
||||
" if len(relevant_rows) == 1:\n",
|
||||
" output_df = pd.concat([output_df, relevant_rows])\n",
|
||||
" # Otherwise we apply the 5 pager logic\n",
|
||||
" elif len(relevant_rows) > 1:\n",
|
||||
" # We use this max page count of all the matching files to compare against others\n",
|
||||
" max_page_count = max(relevant_rows['Page Count'])\n",
|
||||
" flag = False\n",
|
||||
" for pgcnt in relevant_rows['Page Count'].unique():\n",
|
||||
" # If we find another matching file within 5 pages of the max count, we can add that to our output\n",
|
||||
" if max_page_count - pgcnt <= 5:\n",
|
||||
" output_df = pd.concat([output_df, relevant_rows[relevant_rows['Page Count'] == pgcnt]])\n",
|
||||
" else:\n",
|
||||
" print(file)\n",
|
||||
"\n",
|
||||
"# here we do some post processing to ensure we are not keeping full duplicate rows and then convert the ouput df to a csv\n",
|
||||
"output_df = output_df.drop_duplicates()\n",
|
||||
"output_df.reset_index(drop=True, inplace=True)\n",
|
||||
"output_df.to_csv('batch6_16_output_file_2.csv')\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": ".venv",
|
||||
"language": "python",
|
||||
"name": "python3"
|
||||
},
|
||||
"language_info": {
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"file_extension": ".py",
|
||||
"mimetype": "text/x-python",
|
||||
"name": "python",
|
||||
"nbconvert_exporter": "python",
|
||||
"pygments_lexer": "ipython3",
|
||||
"version": "3.12.3"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 2
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
import boto3
|
||||
import csv
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
def list_s3_files(bucket_name, folder):
|
||||
s3 = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
paginator = s3.get_paginator('list_objects_v2')
|
||||
operation_parameters = {'Bucket': bucket_name, 'Prefix': folder}
|
||||
file_names = set()
|
||||
|
||||
for page in paginator.paginate(**operation_parameters):
|
||||
if 'Contents' in page:
|
||||
for content in page['Contents']:
|
||||
file_name = content['Key']
|
||||
if not file_name.endswith('/'):
|
||||
file_name_without_extension = file_name.replace(folder, '', 1)[:-4]
|
||||
# file_name_without_extension = file_name[:-4]
|
||||
file_names.add(file_name_without_extension)
|
||||
|
||||
return file_names
|
||||
|
||||
def list_local_files(path):
|
||||
files = set()
|
||||
try:
|
||||
for _, _, filenames in os.walk(path):
|
||||
for filename in filenames:
|
||||
files.add(filename[:-4])
|
||||
except Exception as e:
|
||||
print(f"Error accessing directory {path}: {e}")
|
||||
return files
|
||||
|
||||
def compare_s3_folders(bucket_name, source1, folder1, source2, folder2, output_csv):
|
||||
|
||||
if source1 == 's3':
|
||||
files_in_folder1 = list_s3_files(bucket_name, folder1)
|
||||
elif source1 == 'local':
|
||||
files_in_folder1 = list_local_files(folder1)
|
||||
elif source1 == 'csv':
|
||||
df = pd.read_csv(folder1,encoding='utf-8')
|
||||
# df['File Name'] = df['File Name'][:-4]
|
||||
files_in_folder1 = set(df['File Name'].to_list())
|
||||
elif source1 == 'xlsb':
|
||||
with pd.ExcelFile(folder1, engine='pyxlsb') as xlsb:
|
||||
first_sheet = xlsb.sheet_names[0]
|
||||
df = xlsb.parse(first_sheet)
|
||||
files_in_folder1 = set(df['Contract Name'].to_list())
|
||||
|
||||
if source2 == 's3':
|
||||
files_in_folder2 = list_s3_files(bucket_name, folder2)
|
||||
elif source2 == 'local':
|
||||
files_in_folder2 = list_local_files(folder2)
|
||||
elif source2 == 'csv':
|
||||
df = pd.read_csv(folder2,encoding='utf-8')
|
||||
# df['File Name'] = df['File Name'][:-4]
|
||||
files_in_folder2 = set(df['File Name'].to_list())
|
||||
elif source2 == 'xlsb':
|
||||
with pd.ExcelFile(folder2, engine='pyxlsb') as xlsb:
|
||||
first_sheet = xlsb.sheet_names[0]
|
||||
df = xlsb.parse(first_sheet)
|
||||
files_in_folder2 = set(df['Contract Name'].to_list())
|
||||
|
||||
|
||||
common_files = files_in_folder1 & files_in_folder2
|
||||
only_in_folder1 = files_in_folder1 - files_in_folder2
|
||||
only_in_folder2 = files_in_folder2 - files_in_folder1
|
||||
|
||||
with open(output_csv, 'w', newline='', encoding='utf-8') as csvfile:
|
||||
csv_writer = csv.writer(csvfile)
|
||||
csv_writer.writerow(['Common Files', f'Only in {folder1}', f'Only in {folder2}'])
|
||||
|
||||
max_length = max(len(common_files), len(only_in_folder1), len(only_in_folder2))
|
||||
for i in range(max_length):
|
||||
row = [
|
||||
list(common_files)[i] if i < len(common_files) and list(common_files)[i] else '',
|
||||
list(only_in_folder1)[i] if i < len(only_in_folder1) and list(only_in_folder1)[i] else '',
|
||||
list(only_in_folder2)[i] if i < len(only_in_folder2) and list(only_in_folder2)[i] else ''
|
||||
]
|
||||
csv_writer.writerow(row)
|
||||
|
||||
bucket_name = 'centene-national-contracting-files'
|
||||
source1_type = 'local' # local / s3 / csv / xlsb
|
||||
source1 = 'Batch 6 TXT Files'
|
||||
source2_type = 'csv' # local / s3 / csv / xlsb
|
||||
source2 = 'Batch 6 Outputs'
|
||||
output_csv = 'batch6/batch6_outputs_diff_with_tracker.csv'
|
||||
|
||||
compare_s3_folders(bucket_name, source1_type, source1, source2_type, source2, output_csv)
|
||||
print(f'Comparison results have been written to {output_csv}')
|
||||
@@ -0,0 +1,37 @@
|
||||
import os
|
||||
import csv
|
||||
import glob
|
||||
from PyPDF2 import PdfReader
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
def get_pdf_info(pdf_path):
|
||||
try:
|
||||
with open(pdf_path, 'rb') as file:
|
||||
pdf_reader = PdfReader(file)
|
||||
page_count = len(pdf_reader.pages)
|
||||
except Exception as e:
|
||||
print(f"Error reading {pdf_path}: {e}")
|
||||
page_count = None
|
||||
return os.path.basename(pdf_path), pdf_path, page_count
|
||||
|
||||
def collect_pdf_info(directory, output_csv):
|
||||
pdf_files = glob.glob(os.path.join(directory, '**', '*.pdf'), recursive=True)
|
||||
|
||||
with open(output_csv, mode='w', newline='', encoding='utf-8') as csv_file:
|
||||
csv_writer = csv.writer(csv_file)
|
||||
csv_writer.writerow(['File Name', 'File Path', 'Page Count'])
|
||||
|
||||
with ThreadPoolExecutor() as executor:
|
||||
|
||||
futures = {executor.submit(get_pdf_info, pdf): pdf for pdf in pdf_files}
|
||||
|
||||
for future in as_completed(futures):
|
||||
result = future.result()
|
||||
if result:
|
||||
csv_writer.writerow(result)
|
||||
|
||||
print(f"CSV file '{output_csv}' created successfully.")
|
||||
|
||||
directory = 'T:/AArete Client Work/Doczy-Production/Restricted/icertis_contracts'
|
||||
output_csv = 'icertis/icertis_files.csv'
|
||||
collect_pdf_info(directory, output_csv)
|
||||
@@ -0,0 +1,62 @@
|
||||
import os
|
||||
import pandas as pd
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from collections import defaultdict
|
||||
import csv
|
||||
|
||||
def build_file_cache(base_directory):
|
||||
file_cache = defaultdict(list)
|
||||
|
||||
for root, dirs, files in os.walk(base_directory):
|
||||
for file in files:
|
||||
file_cache[file].append(os.path.join(root, file))
|
||||
|
||||
return file_cache
|
||||
|
||||
def find_files(file_cache, filenames):
|
||||
found_files = {}
|
||||
for filename in filenames:
|
||||
if filename in file_cache:
|
||||
found_files[filename] = file_cache[filename][0]
|
||||
else:
|
||||
found_files[filename] = "Not Found"
|
||||
return found_files
|
||||
|
||||
def parallel_search(base_directory, filenames, max_workers=50):
|
||||
print("Building file cache...")
|
||||
file_cache = build_file_cache(base_directory)
|
||||
with open('all_file_paths.csv', mode='w', newline='', encoding='utf-8') as csv_file:
|
||||
writer = csv.writer(csv_file)
|
||||
for key, values in file_cache.items():
|
||||
row = [key] + values if isinstance(values, list) else [key, values]
|
||||
writer.writerow(row)
|
||||
print(f"Dictionary has been successfully written to 'all_file_paths.csv'.")
|
||||
print(f"File cache built with {len(file_cache)} unique files.")
|
||||
found_files = {}
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
futures = {executor.submit(find_files, file_cache, chunk): chunk
|
||||
for chunk in chunked(filenames, len(filenames) // max_workers)}
|
||||
|
||||
for future in as_completed(futures):
|
||||
found_files.update(future.result())
|
||||
|
||||
return found_files
|
||||
|
||||
def chunked(iterable, n):
|
||||
for i in range(0, len(iterable), n):
|
||||
yield iterable[i:i + n]
|
||||
|
||||
def search_files_from_csv(csv_file, base_directory, output_csv):
|
||||
df = pd.read_csv(csv_file)
|
||||
df['filename'] += ".Pdf"
|
||||
filenames = df['filename'].tolist()
|
||||
print(f"Searching for {len(filenames)} files in '{base_directory}'...")
|
||||
file_paths = parallel_search(base_directory, filenames)
|
||||
df['file_path'] = df['filename'].apply(lambda x: file_paths.get(x, "Not Found"))
|
||||
df.to_csv(output_csv, index=False)
|
||||
print(f"Results saved to '{output_csv}'.")
|
||||
|
||||
input_csv = "missing_files_renaming.csv"
|
||||
base_dir = "T:/AArete Client Work/Doczy-Production/Restricted/2024-06-28-pdf/"
|
||||
output_csv = "missing_file_paths_2.csv"
|
||||
search_files_from_csv(input_csv, base_dir, output_csv)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,60 @@
|
||||
import boto3
|
||||
import concurrent.futures
|
||||
|
||||
s3 = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def move_file(bucket_name, src_folder, dest_folder, file_key):
|
||||
src_file_path = f"{src_folder}/{file_key}"
|
||||
dest_file_path = f"{dest_folder}/{file_key}"
|
||||
|
||||
try:
|
||||
s3.copy_object(
|
||||
Bucket=bucket_name,
|
||||
CopySource={'Bucket': bucket_name, 'Key': src_file_path},
|
||||
Key=dest_file_path
|
||||
)
|
||||
|
||||
s3.delete_object(Bucket=bucket_name, Key=src_file_path)
|
||||
print(f"Moved {src_file_path} to {dest_file_path}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error moving {src_file_path}: {e}")
|
||||
|
||||
def list_files(bucket_name, folder, extension=None):
|
||||
files = []
|
||||
paginator = s3.get_paginator('list_objects_v2')
|
||||
|
||||
try:
|
||||
for page in paginator.paginate(Bucket=bucket_name, Prefix=folder):
|
||||
contents = page.get('Contents', [])
|
||||
for item in contents:
|
||||
key = item['Key']
|
||||
if not extension or key.endswith(extension):
|
||||
files.append(key.split('/')[-1])
|
||||
except Exception as e:
|
||||
print(f"Error listing files in {bucket_name}/{folder}: {e}")
|
||||
|
||||
return files
|
||||
|
||||
def move_files_multi_thread(bucket_name, src_folder, dest_folder, num_threads=10):
|
||||
files = list_files(bucket_name, src_folder)
|
||||
|
||||
if not files:
|
||||
print(f"No {extension} files found in the source folder.")
|
||||
return
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
|
||||
futures = [
|
||||
executor.submit(move_file, bucket_name, src_folder, dest_folder, file_key)
|
||||
for file_key in files
|
||||
]
|
||||
|
||||
concurrent.futures.wait(futures)
|
||||
print(f"All {extension} files have been moved.")
|
||||
|
||||
bucket_name = 'centene-national-contracting-files'
|
||||
src_folder = 'batch6_16_file/redundant_files'
|
||||
dest_folder = 'batch6_16_file/txt_files'
|
||||
extension = '.txt'
|
||||
|
||||
move_files_multi_thread(bucket_name, src_folder, dest_folder, num_threads=40)
|
||||
@@ -0,0 +1,54 @@
|
||||
import csv
|
||||
import boto3
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
s3 = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def list_all_files(bucket_name):
|
||||
all_keys = []
|
||||
try:
|
||||
paginator = s3.get_paginator('list_objects_v2')
|
||||
for page in paginator.paginate(Bucket=bucket_name):
|
||||
if 'Contents' in page:
|
||||
all_keys.extend([item['Key'] for item in page['Contents']])
|
||||
except ClientError as e:
|
||||
print(f"Error listing files in bucket {bucket_name}: {e}")
|
||||
return all_keys
|
||||
|
||||
def check_file_in_bucket(filename, all_keys):
|
||||
locations = [key for key in all_keys if (key.endswith(filename))]
|
||||
if copy:
|
||||
for l in locations:
|
||||
s3.copy_object(
|
||||
CopySource={'Bucket': bucket_name, 'Key': l},
|
||||
Bucket=bucket_name,
|
||||
Key=destination_prefix + "/" + filename
|
||||
)
|
||||
exists = bool(locations)
|
||||
return filename, exists, locations
|
||||
|
||||
def process_csv(input_csv, output_csv, bucket_name):
|
||||
|
||||
with open(input_csv, mode='r', newline='', encoding='utf8') as infile, open(output_csv, mode='w', newline='', encoding='utf8') as outfile:
|
||||
reader = csv.reader(infile)
|
||||
writer = csv.writer(outfile)
|
||||
writer.writerow(['File Name', 'exists', 'locations']) # Write header
|
||||
|
||||
with ThreadPoolExecutor() as executor:
|
||||
futures = [executor.submit(check_file_in_bucket, row[0] + file_extension, all_keys) for row in reader]
|
||||
|
||||
for future in futures:
|
||||
filename, exists, locations = future.result()
|
||||
writer.writerow([filename, exists, locations])
|
||||
|
||||
input_csv = 'batch11/batch11_missing_files.csv'
|
||||
output_csv = 'batch11/batch11_missing_files_movement_confirmation.csv'
|
||||
bucket_name = 'centene-national-contracting-files'
|
||||
destination_prefix = 'batch_11_priority_files/diff_files_v2'
|
||||
copy = True
|
||||
file_extension = '.txt'
|
||||
|
||||
all_keys = list_all_files(bucket_name)
|
||||
process_csv(input_csv, output_csv, bucket_name)
|
||||
print("CSV processing completed.")
|
||||
@@ -0,0 +1,68 @@
|
||||
import boto3
|
||||
import pandas as pd
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
s3_client = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def get_filenames_from_s3(bucket, prefix):
|
||||
paginator = s3_client.get_paginator('list_objects_v2')
|
||||
page_iterator = paginator.paginate(Bucket=bucket, Prefix=prefix)
|
||||
|
||||
filenames = set()
|
||||
for page in page_iterator:
|
||||
if 'Contents' in page:
|
||||
for obj in page['Contents']:
|
||||
filenames.add((obj['Key'].split('/')[-1])[:-4])
|
||||
|
||||
return filenames
|
||||
|
||||
def copy_file_if_exists(file_name, source_bucket, source_prefix, destination_bucket, destination_prefix, existing_files):
|
||||
if file_name in existing_files:
|
||||
source_key = f"{source_prefix}/{file_name}".strip('/') + ".pdf"
|
||||
destination_key = f"{destination_prefix}/{file_name}".strip('/') + ".pdf"
|
||||
|
||||
# s3_client.copy_object(
|
||||
# CopySource={'Bucket': source_bucket, 'Key': source_key},
|
||||
# Bucket=destination_bucket,
|
||||
# Key=destination_key
|
||||
# )
|
||||
|
||||
response = s3_client.get_object(Bucket=source_bucket, Key=source_key)
|
||||
file_content = response['Body'].read()
|
||||
# print(f"Downloaded {source_key} from s3://{source_bucket}")
|
||||
s3_client.put_object(Bucket=destination_bucket, Key=destination_key, Body=file_content)
|
||||
print(f"Copied: {file_name} to {destination_prefix}")
|
||||
else:
|
||||
print(f"File not found: {file_name} in {source_prefix}")
|
||||
rerun = pd.concat([rerun, df[df['File name Without Extension'] == file_name]])
|
||||
|
||||
def copy_files_multithreaded(file_list, source_bucket, source_prefix, destination_bucket, destination_prefix, existing_files, max_workers=20):
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
futures = [
|
||||
executor.submit(copy_file_if_exists, file_name, source_bucket, source_prefix, destination_bucket, destination_prefix, existing_files)
|
||||
for file_name in file_list
|
||||
]
|
||||
|
||||
for future in as_completed(futures):
|
||||
try:
|
||||
future.result()
|
||||
except Exception as e:
|
||||
print(f"Error copying file: {e}")
|
||||
|
||||
csv_path = 'new_duplicates_in_batch3.csv'
|
||||
file_name_column = 'File Name Without Extension'
|
||||
source_bucket = 'centene-national-contracting-files'
|
||||
search_prefix = 'batch_3_priority_files/txt_files'
|
||||
source_prefix = 'batch_3_priority_files/pdf_files'
|
||||
destination_bucket = 'centene-national-contracting-files'
|
||||
destination_prefix = 'batch_3_priority_files/re_run_files'
|
||||
|
||||
df = pd.read_csv(csv_path)
|
||||
rerun = pd.DataFrame(columns=df.columns)
|
||||
file_names = df[file_name_column].dropna().tolist()
|
||||
print(f"Loaded {len(file_names)} file names from CSV.")
|
||||
existing_files = get_filenames_from_s3(source_bucket, search_prefix)
|
||||
print(f"Found {len(existing_files)} files in S3 source folder '{source_prefix}'.")
|
||||
copy_files_multithreaded(file_names, source_bucket, source_prefix, destination_bucket, destination_prefix, existing_files)
|
||||
rerun.reset_index(drop=True, inplace=True)
|
||||
rerun.to_csv('batch3_rerun.csv')
|
||||
@@ -0,0 +1,49 @@
|
||||
import boto3
|
||||
import re
|
||||
import csv
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
BUCKET_NAME = "centene-national-contracting-files"
|
||||
PREFIX = "batch_1_priority_files/text_files/"
|
||||
OUTPUT_CSV = "batch1/batch1_s3_files_page_counts.csv"
|
||||
|
||||
s3_client = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def process_file(s3_key):
|
||||
try:
|
||||
response = s3_client.get_object(Bucket=BUCKET_NAME, Key=s3_key)
|
||||
content = response['Body'].read().decode('utf-8')
|
||||
|
||||
matches = re.findall(r"Start of Page No\. = (\d+)", content)
|
||||
|
||||
last_page = matches[-1] if matches else None
|
||||
|
||||
return s3_key.split('/')[-1], last_page
|
||||
except Exception as e:
|
||||
print(f"Error processing {s3_key}: {e}")
|
||||
return s3_key.split('/')[-1], None
|
||||
|
||||
def list_s3_objects(bucket, prefix):
|
||||
paginator = s3_client.get_paginator('list_objects_v2')
|
||||
for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
|
||||
for obj in page.get('Contents', []):
|
||||
if obj['Key'].endswith('.txt'):
|
||||
yield obj['Key']
|
||||
|
||||
try:
|
||||
files = list(list_s3_objects(BUCKET_NAME, PREFIX))
|
||||
except Exception as e:
|
||||
print(f"Error listing objects in S3: {e}")
|
||||
|
||||
results = []
|
||||
with ThreadPoolExecutor(max_workers= 50) as executor:
|
||||
futures = [executor.submit(process_file, file) for file in files]
|
||||
for future in futures:
|
||||
results.append(future.result())
|
||||
|
||||
with open(OUTPUT_CSV, mode='w', newline='', encoding='utf-8') as csvfile:
|
||||
writer = csv.writer(csvfile)
|
||||
writer.writerow(["Filename", "Page Count"])
|
||||
writer.writerows(results)
|
||||
|
||||
print(f"Output written to {OUTPUT_CSV}")
|
||||
@@ -0,0 +1,44 @@
|
||||
import boto3
|
||||
import csv
|
||||
|
||||
"""
|
||||
This is a basic script to list all files in a given S3 bucket and prefix.
|
||||
There is need for listing out some file names in a given S3 bucket and prefix as client needs to reconcile the files.
|
||||
There might not be any need for this script. But useful for future reference.
|
||||
"""
|
||||
|
||||
def list_files_in_s3(bucket_name, s3_prefix, profile_name, output_csv):
|
||||
# Initialize a session using a custom profile
|
||||
session = boto3.Session(profile_name=profile_name)
|
||||
s3_client = session.client('s3')
|
||||
|
||||
# List all files in the given S3 bucket and prefix
|
||||
paginator = s3_client.get_paginator('list_objects_v2')
|
||||
operation_parameters = {
|
||||
'Bucket': bucket_name,
|
||||
'Prefix': s3_prefix
|
||||
}
|
||||
|
||||
# Open the CSV file for writing
|
||||
with open(output_csv, mode='w', newline='') as file:
|
||||
writer = csv.writer(file)
|
||||
writer.writerow(['file_name', 'sub_prefix'])
|
||||
|
||||
# Paginate through all files
|
||||
for page in paginator.paginate(**operation_parameters):
|
||||
if 'Contents' in page:
|
||||
for obj in page['Contents']:
|
||||
key = obj['Key']
|
||||
# Extract the sub-prefix by removing the base prefix from the key
|
||||
sub_prefix = key[len(s3_prefix):].split('/')[1] if len(key[len(s3_prefix):].split('/')) > 1 else ''
|
||||
writer.writerow([key, sub_prefix])
|
||||
print(f"Found: {key}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Specify the S3 bucket, prefix, profile name, and output CSV file
|
||||
bucket_name = "centene-texas-files" # Update your S3 bucket name
|
||||
s3_prefix = "bedrock_rerun_text_files/not_executed/" # Update the S3 prefix (folder path in S3)
|
||||
profile_name = "" # Update the AWS CLI profile name
|
||||
output_csv = "not_executed_tx_rerun_list.csv" # Update the output CSV file name
|
||||
|
||||
list_files_in_s3(bucket_name, s3_prefix, profile_name, output_csv)
|
||||
@@ -0,0 +1,182 @@
|
||||
|
||||
|
||||
import pandas as pd
|
||||
import os
|
||||
import boto3
|
||||
|
||||
"""
|
||||
This script is used to match file names between a local CSV and an Excel sheet, and copy the matched files to a specified prefix in an S3 bucket using boto3. The script performs the following main steps:
|
||||
|
||||
1. **Loading Files**: Load all files from the CSV and subset files from the Excel sheet.
|
||||
2. **Normalizing File Names**: Normalize file names for proper matching by standardizing formats.
|
||||
3. **Matching Files**: Identify matching and missing files between the two data sources.
|
||||
4. **Copying Matched Files**: Copy matched files to a specific prefix in an S3 bucket.
|
||||
|
||||
The script uses multiple sheets in an Excel file, processes each one, and handles S3 operations for moving matched files.
|
||||
"""
|
||||
|
||||
# -------------------------------
|
||||
# Configuration Variables
|
||||
# -------------------------------
|
||||
|
||||
# File paths
|
||||
# s3://centene-texas-files/missc-tx-files/all_files.csv
|
||||
all_files_path = 'C:\\Doczy\\Centene Texas\\all_files.csv' # Path to the CSV containing all file names
|
||||
excel_path = 'additional-9k.xlsx' # Path to the Excel file containing subset file names from client
|
||||
sheet_names = ['Sheet1'] # List of sheet names to be processed. This depends on what the client team shares
|
||||
|
||||
# AWS Configuration
|
||||
profile_name = '' # AWS CLI profile name for authentication
|
||||
bucket_name = 'centene-texas-files' # S3 bucket name. This script is mainly used only with Texas files
|
||||
|
||||
# Destination prefixes in S3
|
||||
new_prefix_map = {
|
||||
'Sheet1': 'bedrock_rerun_text_files/not_executed/rerun_20241125/'
|
||||
|
||||
# WE CAN USE THIS MAP IF THERE ARE MULTIPLE SHEETS TO BE PROCESSED
|
||||
|
||||
# 'Missing Row': 'bedrock_rerun_text_files/not_executed/b_missing_row_output/',
|
||||
# 'IRS #': 'bedrock_rerun_text_files/not_executed/a_c_blank_irs/'
|
||||
}
|
||||
|
||||
# -------------------------------
|
||||
# Functions
|
||||
# -------------------------------
|
||||
|
||||
def load_files(all_files_path, excel_path, sheet_name):
|
||||
"""
|
||||
Loads all files from CSV and subset files from Excel sheet.
|
||||
|
||||
:param all_files_path: Path to the CSV file containing all file names
|
||||
:param excel_path: Path to the Excel file containing subset file names
|
||||
:param sheet_name: Sheet name in the Excel file to load
|
||||
:return: DataFrames for all files and subset files
|
||||
"""
|
||||
all_files_df = pd.read_csv(all_files_path) # Load all files from CSV
|
||||
subset_files_df = pd.read_excel(excel_path, sheet_name=sheet_name, header=1) # Load subset files from specified Excel sheet
|
||||
return all_files_df, subset_files_df
|
||||
|
||||
def normalize_file_names(all_files_df, subset_files_df):
|
||||
"""
|
||||
Normalizes file names in both DataFrames for matching.
|
||||
|
||||
:param all_files_df: DataFrame containing all file names
|
||||
:param subset_files_df: DataFrame containing subset file names
|
||||
:return: Normalized DataFrames for all files and subset files
|
||||
"""
|
||||
# Append '.txt' to CONTRACT NAME in the subset files DataFrame to create a consistent file name format
|
||||
subset_files_df['File Name '] = subset_files_df['CONTRACT NAME'].apply(lambda x: x + '.txt')
|
||||
|
||||
# Remove leading hyphen from base_name if present and convert to lowercase for normalization
|
||||
all_files_df['base_name'] = all_files_df['base_name'].fillna('').apply(lambda x: x[1:] if isinstance(x, str) and x.startswith('-') else x)
|
||||
all_files_df['base_name'] = all_files_df['base_name'].str.lower()
|
||||
|
||||
# Convert file names in subset DataFrame to lowercase for normalization
|
||||
subset_files_df['File Name '] = subset_files_df['File Name '].str.lower()
|
||||
|
||||
return all_files_df, subset_files_df
|
||||
|
||||
def match_files(all_files_df, subset_files_df):
|
||||
"""
|
||||
Matches files between the all files DataFrame and the subset files DataFrame.
|
||||
|
||||
:param all_files_df: DataFrame containing all file names
|
||||
:param subset_files_df: DataFrame containing subset file names
|
||||
:return: DataFrames for matched files and missing files
|
||||
"""
|
||||
# Find all files in all_files_df that have a base name matching any file name in subset_files_df
|
||||
matched_files_df = all_files_df[all_files_df['base_name'].isin(subset_files_df['File Name '])]
|
||||
|
||||
# Find all files in subset_files_df that do not have a matching base name in all_files_df
|
||||
missing_files_df = subset_files_df[~subset_files_df['File Name '].isin(all_files_df['base_name'])]
|
||||
|
||||
return matched_files_df, missing_files_df
|
||||
|
||||
def process_multiple_sheets(all_files_path, excel_path, sheet_names):
|
||||
"""
|
||||
Processes multiple sheets from the Excel file and matches files with the CSV.
|
||||
|
||||
:param all_files_path: Path to the CSV file containing all file names
|
||||
:param excel_path: Path to the Excel file containing subset file names
|
||||
:param sheet_names: List of sheet names to be processed
|
||||
:return: Dictionary containing matched and missing files DataFrames for each sheet
|
||||
"""
|
||||
results = {}
|
||||
for sheet in sheet_names:
|
||||
print(f"Processing Sheet: {sheet}")
|
||||
|
||||
# Load all files and subset files for the current sheet
|
||||
all_files_df, subset_files_df = load_files(all_files_path, excel_path, sheet)
|
||||
|
||||
# Normalize file names for both DataFrames
|
||||
all_files_df, subset_files_df = normalize_file_names(all_files_df, subset_files_df)
|
||||
|
||||
# Match files between the two DataFrames
|
||||
matched_files_df, missing_files_df = match_files(all_files_df, subset_files_df)
|
||||
|
||||
# Store the results for each sheet in the dictionary
|
||||
results[sheet] = {
|
||||
'matched_files_df': matched_files_df,
|
||||
'missing_files_df': missing_files_df
|
||||
}
|
||||
|
||||
return results
|
||||
|
||||
def copy_files_to_new_prefix(matched_files_df, bucket_name, destination_prefix):
|
||||
"""
|
||||
Copies matched files to a new prefix in the specified S3 bucket.
|
||||
|
||||
:param matched_files_df: DataFrame containing matched file information
|
||||
:param bucket_name: S3 bucket name
|
||||
:param destination_prefix: Prefix in the S3 bucket to copy files to
|
||||
"""
|
||||
# Create a boto3 session and S3 client
|
||||
session = boto3.Session(profile_name=profile_name)
|
||||
s3 = session.client('s3')
|
||||
|
||||
# Iterate over each matched file and copy it to the new destination prefix
|
||||
for _, row in matched_files_df.iterrows():
|
||||
source_key = row['file_name'] # Source key in S3 bucket
|
||||
destination_key = destination_prefix + os.path.basename(source_key) # Destination key with new prefix
|
||||
print(f"Copying {source_key} to {destination_key}")
|
||||
try:
|
||||
# Define the copy source and perform the copy operation
|
||||
copy_source = {'Bucket': bucket_name, 'Key': source_key}
|
||||
s3.copy(copy_source, bucket_name, destination_key)
|
||||
print(f"Copied {source_key} to {destination_key}")
|
||||
except Exception as e:
|
||||
# Handle any exceptions that occur during the copy process
|
||||
print(f"FILE NOT FOUND: {e}")
|
||||
|
||||
# -------------------------------
|
||||
# Main Script
|
||||
# -------------------------------
|
||||
|
||||
# Process multiple sheets and get results
|
||||
results = process_multiple_sheets(all_files_path, excel_path, sheet_names)
|
||||
|
||||
# Combine matched and missing files from all sheets
|
||||
combined_df = pd.DataFrame() # DataFrame to store all matched files
|
||||
missing_files_df = pd.DataFrame() # DataFrame to store all missing files
|
||||
for sheet, data in results.items():
|
||||
print(f'{sheet}:')
|
||||
print(f'Matched files: {data["matched_files_df"].shape[0]}') # Print the number of matched files for the sheet
|
||||
print(f'Missing files: {data["missing_files_df"].shape[0]}') # Print the number of missing files for the sheet
|
||||
print(data['missing_files_df']['CONTRACT NAME'].values) # Print the names of the missing files
|
||||
|
||||
# Append matched and missing files to the combined DataFrames
|
||||
combined_df = pd.concat([combined_df, data['matched_files_df']], ignore_index=True)
|
||||
missing_files_df = pd.concat([missing_files_df, data['missing_files_df']], ignore_index=True)
|
||||
|
||||
# Copy matched files to the appropriate prefix in S3
|
||||
for sheet, data in results.items():
|
||||
matched_files_df = data['matched_files_df'] # DataFrame containing matched files for the sheet
|
||||
destination_prefix = new_prefix_map.get(sheet) # Get the destination prefix for the sheet
|
||||
|
||||
if destination_prefix:
|
||||
# Copy matched files to the new prefix in S3
|
||||
copy_files_to_new_prefix(matched_files_df, bucket_name, destination_prefix)
|
||||
else:
|
||||
print(f"No destination prefix found for sheet: {sheet}")
|
||||
|
||||
print("File copying complete.")
|
||||
@@ -0,0 +1,173 @@
|
||||
import boto3
|
||||
import pandas as pd
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
"""
|
||||
This script is a work in progress.
|
||||
TODO:
|
||||
1. Read the DS batch tracker and get the total document counts per client and hence calculate the total bedrock cost per client.
|
||||
|
||||
This script is used to automate the process of fetching AWS costs for specific services and additional charges, as well as counting the number of records in each client's S3 bucket.
|
||||
The script performs the following main steps:
|
||||
|
||||
1. **Get Service Costs**: Fetch costs for specified AWS services using the AWS Cost Explorer API.
|
||||
2. **Get Costs by Record Type**: Fetch additional charges (e.g., Tax, Support) using the AWS Cost Explorer API.
|
||||
3. **Process Bucket**: Count the number of records in each client's S3 bucket for a specified date range.
|
||||
4. **Export Data**: Export the cost data and bucket record counts to CSV files.
|
||||
|
||||
"""
|
||||
|
||||
def get_service_costs(start_date, end_date, services, profile_name):
|
||||
client = boto3.Session(profile_name=profile_name).client('ce') # Cost Explorer client
|
||||
response = client.get_cost_and_usage(
|
||||
TimePeriod={
|
||||
'Start': start_date,
|
||||
'End': end_date # AWS Cost Explorer 'End' date is exclusive
|
||||
},
|
||||
Granularity='DAILY',
|
||||
Metrics=['UnblendedCost'],
|
||||
GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}],
|
||||
Filter={
|
||||
'Dimensions': {
|
||||
'Key': 'SERVICE',
|
||||
'Values': services
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
costs = {}
|
||||
for result in response['ResultsByTime']:
|
||||
for group in result.get('Groups', []):
|
||||
service = group['Keys'][0]
|
||||
amount = float(group['Metrics']['UnblendedCost']['Amount'])
|
||||
costs[service] = costs.get(service, 0) + amount
|
||||
return costs
|
||||
|
||||
def get_costs_by_record_type(start_date, end_date, record_types, profile_name):
|
||||
client = boto3.Session(profile_name=profile_name).client('ce')
|
||||
response = client.get_cost_and_usage(
|
||||
TimePeriod={'Start': start_date, 'End': end_date},
|
||||
Granularity='DAILY',
|
||||
Metrics=['UnblendedCost'],
|
||||
GroupBy=[{'Type': 'DIMENSION', 'Key': 'RECORD_TYPE'}],
|
||||
Filter={
|
||||
'Dimensions': {
|
||||
'Key': 'RECORD_TYPE',
|
||||
'Values': record_types
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
costs = {}
|
||||
for result in response['ResultsByTime']:
|
||||
for group in result.get('Groups', []):
|
||||
record_type = group['Keys'][0]
|
||||
amount = float(group['Metrics']['UnblendedCost']['Amount'])
|
||||
costs[record_type] = costs.get(record_type, 0) + amount
|
||||
return costs
|
||||
|
||||
def process_bucket(bucket_name, start_date_obj, end_date_obj, profile_name):
|
||||
s3 = boto3.Session(profile_name=profile_name).client('s3')
|
||||
prefix = 'processed_batch_ids/'
|
||||
paginator = s3.get_paginator('list_objects_v2')
|
||||
page_iterator = paginator.paginate(Bucket=bucket_name, Prefix=prefix)
|
||||
|
||||
total_records = 0
|
||||
for page in page_iterator:
|
||||
for obj in page.get('Contents', []):
|
||||
key = obj['Key']
|
||||
last_modified = obj['LastModified']
|
||||
if key.endswith('.csv') and start_date_obj <= last_modified.date() <= end_date_obj:
|
||||
# Read the CSV file into a dataframe
|
||||
csv_obj = s3.get_object(Bucket=bucket_name, Key=key)
|
||||
df = pd.read_csv(csv_obj['Body'])
|
||||
total_records += len(df)
|
||||
|
||||
# For drilling down
|
||||
# print(f"Number of records in {key}: {len(df)}")
|
||||
return total_records
|
||||
|
||||
def main():
|
||||
# Input parameters
|
||||
start_date = '2024-11-01' # Replace with your start date
|
||||
end_date = '2024-11-25' # Replace with your end date
|
||||
client_buckets = ['highmark-files', 'la-care-files',
|
||||
'centene-fidelis-files',
|
||||
'centene-healthnet-files',
|
||||
'centene-national-contracting-files',
|
||||
'centene-texas-files',
|
||||
'bcbs-files',
|
||||
'community-health-choice-files',
|
||||
'humana-files',
|
||||
'scan-health-files',
|
||||
'texas-children-files',
|
||||
'village-care']
|
||||
profile_name = 'doczy_uat'
|
||||
|
||||
# List of services to fetch costs for
|
||||
services = [
|
||||
'Amazon Textract',
|
||||
'Amazon Bedrock',
|
||||
'Claude Instant (Amazon Bedrock Edition)',
|
||||
'Claude 3 Haiku (Amazon Bedrock Edition)',
|
||||
'Claude 3.5 Sonnet (Amazon Bedrock Edition)'
|
||||
]
|
||||
|
||||
# Convert date strings to datetime objects
|
||||
start_date_obj = datetime.strptime(start_date, '%Y-%m-%d').date()
|
||||
end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date()
|
||||
|
||||
# Adjust end_date for AWS Cost Explorer API (end date is exclusive)
|
||||
end_date_ce = (datetime.strptime(end_date, '%Y-%m-%d') + timedelta(days=1)).strftime('%Y-%m-%d')
|
||||
|
||||
# Get costs for specified services
|
||||
service_costs = get_service_costs(start_date, end_date_ce, services, profile_name)
|
||||
|
||||
# Get charges for Tax and Support
|
||||
tax_costs = get_costs_by_record_type(start_date, end_date_ce, ['Tax'], profile_name)
|
||||
support_costs = get_costs_by_record_type(start_date, end_date_ce, ['Support'], profile_name)
|
||||
|
||||
# Prepare data for CSV export
|
||||
cost_data = []
|
||||
|
||||
# Output costs
|
||||
print("AWS Service Costs:")
|
||||
for service in services:
|
||||
cost = service_costs.get(service, 0.0)
|
||||
print(f" {service}: ${cost:.2f}")
|
||||
cost_data.append({'Category': 'Service', 'Item': service, 'Cost': cost})
|
||||
|
||||
print("\nAdditional Charges:")
|
||||
tax_cost = tax_costs.get('Tax', 0.0)
|
||||
support_cost = support_costs.get('Support', 0.0)
|
||||
print(f" Tax: ${tax_cost:.2f}")
|
||||
print(f" Support: ${support_cost:.2f}")
|
||||
cost_data.append({'Category': 'Additional Charges', 'Item': 'Tax', 'Cost': tax_cost})
|
||||
cost_data.append({'Category': 'Additional Charges', 'Item': 'Support', 'Cost': support_cost})
|
||||
|
||||
total_costs = sum([cost['Cost'] for cost in cost_data])
|
||||
cost_data.append({'Category': 'Total', 'Item': 'Total', 'Cost': total_costs})
|
||||
|
||||
# Process each bucket and count records
|
||||
print("\nBucket-wise Record Counts:")
|
||||
bucket_data = []
|
||||
for bucket in client_buckets:
|
||||
total_records = process_bucket(bucket, start_date_obj, end_date_obj, profile_name)
|
||||
print(f" {bucket}: {total_records} records")
|
||||
bucket_data.append({'Bucket': bucket, 'Records': total_records})
|
||||
|
||||
# Option to export data as CSV
|
||||
export_csv = True # Set to True to export CSV
|
||||
if export_csv:
|
||||
# Export cost data
|
||||
cost_df = pd.DataFrame(cost_data)
|
||||
cost_df.to_csv('aws_costs.csv', index=False)
|
||||
print("\nService costs exported to 'aws_costs.csv'.")
|
||||
|
||||
# Export bucket data
|
||||
bucket_df = pd.DataFrame(bucket_data)
|
||||
bucket_df.to_csv('bucket_records.csv', index=False)
|
||||
print("Bucket record counts exported to 'bucket_records.csv'.")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,64 @@
|
||||
import boto3
|
||||
import pandas as pd
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
|
||||
"""
|
||||
This script is used to copy complex contract text files to a different folder in the same S3 bucket.
|
||||
The script reads a CSV file that contains the filenames and their table counts. It filters the rows where the table count is greater than or equal to 1.
|
||||
|
||||
This script is NOT CLIENT SPECIFIC. It is a generic script that can be used for any client.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# Function to copy a single file
|
||||
def copy_file(bucket_name, source_prefix, dest_prefix, file_name, profile_name):
|
||||
try:
|
||||
# Initialize S3 client
|
||||
session = boto3.Session(profile_name=profile_name)
|
||||
s3 = session.client('s3')
|
||||
copy_source = {'Bucket': bucket_name, 'Key': f'{source_prefix}/{file_name}'}
|
||||
dest_key = f'{dest_prefix}/{file_name}'
|
||||
s3.copy_object(CopySource=copy_source, Bucket=bucket_name, Key=dest_key)
|
||||
print(f'Successfully copied from {copy_source} to {dest_key}')
|
||||
except Exception as e:
|
||||
# print(f'Failed to copy {file_name}: {e}')
|
||||
print('File not in bucket. Skipping')
|
||||
pass
|
||||
|
||||
# Function to process copying in parallel
|
||||
def copy_files_in_parallel(bucket_name, source_prefix, dest_prefix, file_list,profile_name ,max_workers=10):
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
futures = [
|
||||
executor.submit(copy_file, bucket_name, source_prefix, dest_prefix, file_name, profile_name=profile_name)
|
||||
for file_name in file_list
|
||||
]
|
||||
for future in as_completed(futures):
|
||||
future.result() # This will raise any exceptions that occurred during execution
|
||||
|
||||
# Main function
|
||||
def main():
|
||||
# Client bucket where the text files are stored
|
||||
bucket_name = 'centene-national-contracting-files'
|
||||
|
||||
# This is used to copy complex contract text files to a different folder
|
||||
source_prefix = 'batch6_16_file/txt_files'
|
||||
dest_prefix_3a = 'batch6_16_file/complex_contract_files/txt_files'
|
||||
max_workers = 50
|
||||
profile_name = 'default' # Set this as per your AWS profile
|
||||
|
||||
# Read the table analysis file (from DS code) and read the filenames and their table counts
|
||||
csv_file = 'C:\\Doczy\\National contracting\\Somefolder\\CNC-6to16-Table-Analysis.csv' # Replace with your CSV file path
|
||||
file_df = pd.read_csv(csv_file)
|
||||
|
||||
# Filter rows where 'tables' value is greater than or equal to 1
|
||||
complex_table_files = file_df[file_df['Table Count'] >= 1]
|
||||
|
||||
# Assuming the CSV has a column 'file_name' with the list of file names
|
||||
file_list = complex_table_files['Filename_pdf'].tolist()
|
||||
|
||||
copy_files_in_parallel(bucket_name, source_prefix, dest_prefix_3a, file_list, profile_name = profile_name, max_workers=max_workers)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,308 @@
|
||||
import csv
|
||||
import boto3
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
import os
|
||||
|
||||
"""
|
||||
This is an adhoc script that is used tot copy files from one prefix to another prefix within the same S3 bucket.
|
||||
|
||||
TYPICALLY USED FOR CENTENE-HEALTHNET
|
||||
|
||||
The need for this script is when the client team requests re-run and the previous runs are stored in a different prefixes.
|
||||
|
||||
The script reads a CSV file containing 'File Name' and 'Type' columns. It then searches for these files in the specified prefixes
|
||||
and copies them to a new destination prefix within the same bucket.
|
||||
|
||||
The script also writes a CSV file containing the files that were not found in the specified prefixes.
|
||||
|
||||
The script is multithreaded to improve performance when searching for files in multiple prefixes.
|
||||
|
||||
This script can be used to just location files by commenting out the copy functionality.
|
||||
|
||||
Usage:
|
||||
1. Update the configuration section below with the appropriate values.
|
||||
2. Run the script using Python 3.
|
||||
|
||||
Note:
|
||||
- Ensure that the CSV file contains 'File Name' and 'Type' columns.
|
||||
- Ensure you have the required permissions to read from and write to the specified S3 bucket.
|
||||
|
||||
"""
|
||||
|
||||
# --------------------- Configuration ---------------------
|
||||
|
||||
# Path to your CSV file from client / DS team that need to be staged
|
||||
CSV_FILE = 'C:\\Doczy\\Health net\\no_tin_filenames.csv' # Ensure this file has 'File Name' and 'Type' columns
|
||||
# AWS Configuration
|
||||
PROFILE_NAME = 'doczy_uat' # Replace with your AWS profile name
|
||||
BUCKET_NAME = 'centene-healthnet-files' # Replace with your S3 bucket name
|
||||
|
||||
# List of prefixes to search within the bucket
|
||||
PREFIXES = [
|
||||
'facility_rerun_092524/txt_files/',
|
||||
'facility-remaining-text-files/',
|
||||
'agreements/text_files/',
|
||||
'ammendments/text_files/',
|
||||
'1kfacilityrun/',
|
||||
'ancillary_1200_rerun/txt_files/',
|
||||
'facility_remaining_1k_rerun/',
|
||||
'facility-table-issue-files/',
|
||||
'facility-test-files/',
|
||||
'facility-textract-rerun/txt_files/',
|
||||
'HN-10-Ancillary-Mix/',
|
||||
'lessthan200pagesfiles/',
|
||||
'new_prof_and_ancillary_1200/',
|
||||
'professional_text_files_for_llm/',
|
||||
'text_files_for_llm/',
|
||||
'professional_paragraph_format_rerun/',
|
||||
'HN-10-Ancillary-Mix/',
|
||||
'facility_rerun_092524/',
|
||||
'adhoc/'
|
||||
# Add more prefixes as needed
|
||||
]
|
||||
|
||||
# Destination prefix for copying files (within the same bucket)
|
||||
DESTINATION_PREFIX = 'facility_rerun_112124/no_tin/' # Replace with your desired destination prefix
|
||||
|
||||
# Number of threads for multithreading
|
||||
MAX_WORKERS = 100
|
||||
|
||||
# Output CSV for files not found
|
||||
NOT_FOUND_CSV = 'C:\\Doczy\\Health net\\files_not_found_no_tin.csv'
|
||||
|
||||
|
||||
# --------------------- Functions ---------------------
|
||||
|
||||
def create_s3_client(profile_name):
|
||||
"""
|
||||
Create a Boto3 S3 client using the specified AWS profile.
|
||||
"""
|
||||
session = boto3.Session(profile_name=profile_name)
|
||||
s3_client = session.client('s3')
|
||||
return s3_client
|
||||
|
||||
def normalize_string(s):
|
||||
"""
|
||||
Normalize a string by removing spaces and converting to lowercase.
|
||||
|
||||
Parameters:
|
||||
s (str): The string to normalize.
|
||||
|
||||
Returns:
|
||||
str: The normalized string.
|
||||
"""
|
||||
return ''.join(s.split()).lower()
|
||||
|
||||
def read_csv(csv_file):
|
||||
"""
|
||||
Read the CSV file and extract the list of file names along with their types.
|
||||
|
||||
Parameters:
|
||||
csv_file (str): Path to the CSV file.
|
||||
|
||||
Returns:
|
||||
list of dict: List containing dictionaries with 'File Name' and 'Type'.
|
||||
"""
|
||||
files = []
|
||||
try:
|
||||
with open(csv_file, mode='r', newline='', encoding='utf-8') as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
files.append({
|
||||
'File Name': row['File Name'],
|
||||
'Type': row['Type']
|
||||
})
|
||||
except FileNotFoundError:
|
||||
print(f"Error: The file {csv_file} does not exist.")
|
||||
except KeyError:
|
||||
print("Error: CSV file must contain 'File Name' and 'Type' columns.")
|
||||
except Exception as e:
|
||||
print(f"An unexpected error occurred while reading the CSV: {e}")
|
||||
return files
|
||||
|
||||
def list_objects_in_prefix(s3_client, bucket, prefix):
|
||||
"""
|
||||
List all objects in a given prefix and return a mapping of normalized keys to actual keys.
|
||||
|
||||
Parameters:
|
||||
s3_client (boto3.client): Boto3 S3 client.
|
||||
bucket (str): S3 bucket name.
|
||||
prefix (str): Prefix to list objects from.
|
||||
|
||||
Returns:
|
||||
dict: Mapping of normalized keys to actual S3 keys.
|
||||
"""
|
||||
normalized_key_mapping = {}
|
||||
continuation_token = None
|
||||
while True:
|
||||
try:
|
||||
if continuation_token:
|
||||
response = s3_client.list_objects_v2(
|
||||
Bucket=bucket,
|
||||
Prefix=prefix,
|
||||
ContinuationToken=continuation_token
|
||||
)
|
||||
else:
|
||||
response = s3_client.list_objects_v2(
|
||||
Bucket=bucket,
|
||||
Prefix=prefix
|
||||
)
|
||||
if 'Contents' in response:
|
||||
for obj in response['Contents']:
|
||||
actual_key = obj['Key']
|
||||
normalized_key = normalize_string(os.path.basename(actual_key).replace('.txt', ''))
|
||||
normalized_key_mapping[normalized_key] = actual_key
|
||||
if response.get('IsTruncated'):
|
||||
continuation_token = response.get('NextContinuationToken')
|
||||
else:
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"Error listing objects in prefix {prefix}: {e}")
|
||||
break
|
||||
return normalized_key_mapping
|
||||
|
||||
def build_normalized_key_mapping(s3_client, bucket, prefixes):
|
||||
"""
|
||||
Build a mapping from normalized filenames to actual S3 keys across all specified prefixes.
|
||||
|
||||
Parameters:
|
||||
s3_client (boto3.client): Boto3 S3 client.
|
||||
bucket (str): S3 bucket name.
|
||||
prefixes (list): List of prefixes to search.
|
||||
|
||||
Returns:
|
||||
dict: Mapping of normalized filenames to actual S3 keys.
|
||||
"""
|
||||
normalized_key_mapping = {}
|
||||
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
|
||||
future_to_prefix = {
|
||||
executor.submit(list_objects_in_prefix, s3_client, bucket, prefix): prefix
|
||||
for prefix in prefixes
|
||||
}
|
||||
for future in as_completed(future_to_prefix):
|
||||
prefix = future_to_prefix[future]
|
||||
try:
|
||||
prefix_mapping = future.result()
|
||||
# Handle potential duplicate normalized keys
|
||||
for norm_key, actual_key in prefix_mapping.items():
|
||||
if norm_key in normalized_key_mapping:
|
||||
# This case is has never been hit in the past
|
||||
# It is possible that the same file name is present in multiple prefixes (For Centene National Contracting)
|
||||
# print(f"Warning: Duplicate normalized key '{norm_key}' found in prefix '{prefix}'.")
|
||||
pass
|
||||
else:
|
||||
normalized_key_mapping[norm_key] = actual_key
|
||||
except Exception as exc:
|
||||
print(f"Prefix {prefix} generated an exception: {exc}")
|
||||
return normalized_key_mapping
|
||||
|
||||
def copy_file(s3_client, bucket, source_key, destination_key):
|
||||
"""
|
||||
Copy a file from source_key to destination_key within the same bucket.
|
||||
|
||||
Parameters:
|
||||
s3_client (boto3.client): Boto3 S3 client.
|
||||
bucket (str): S3 bucket name.
|
||||
source_key (str): Source S3 key.
|
||||
destination_key (str): Destination S3 key.
|
||||
"""
|
||||
copy_source = {'Bucket': bucket, 'Key': source_key}
|
||||
try:
|
||||
s3_client.copy(copy_source, bucket, destination_key)
|
||||
print(f"Copied {source_key} to {destination_key}")
|
||||
except Exception as e:
|
||||
print(f"Failed to copy {source_key} to {destination_key}: {e}")
|
||||
|
||||
def write_not_found_csv(not_found_files, output_csv):
|
||||
"""
|
||||
Write the list of not found files to a CSV file.
|
||||
|
||||
Parameters:
|
||||
not_found_files (list of dict): List containing dictionaries with 'File Name' and 'Type'.
|
||||
output_csv (str): Path to the output CSV file.
|
||||
"""
|
||||
try:
|
||||
with open(output_csv, mode='w', newline='', encoding='utf-8') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=['File Name', 'Type'])
|
||||
writer.writeheader()
|
||||
for file in not_found_files:
|
||||
writer.writerow(file)
|
||||
print(f"Successfully wrote {len(not_found_files)} not found files to {output_csv}")
|
||||
except Exception as e:
|
||||
print(f"Failed to write not found files to CSV: {e}")
|
||||
|
||||
def main():
|
||||
# Create S3 client
|
||||
s3_client = create_s3_client(PROFILE_NAME)
|
||||
|
||||
# Read filenames and types from CSV
|
||||
files = read_csv(CSV_FILE)
|
||||
if not files:
|
||||
print("No files to process. Exiting.")
|
||||
return
|
||||
print(f"Total files to search: {len(files)}")
|
||||
|
||||
# Normalize filenames and build a mapping from normalized name to original file info
|
||||
normalized_files_mapping = {}
|
||||
for file in files:
|
||||
original_name = file['File Name']
|
||||
normalized_name = normalize_string(original_name) + '.txt' # Append '.txt' as per original logic
|
||||
normalized_files_mapping[normalized_name] = file
|
||||
|
||||
# Build normalized key mapping from S3
|
||||
print("Listing and mapping S3 objects...")
|
||||
normalized_key_mapping = build_normalized_key_mapping(s3_client, BUCKET_NAME, PREFIXES)
|
||||
print(f"Total S3 objects mapped: {len(normalized_key_mapping)}")
|
||||
|
||||
# Match normalized filenames with normalized S3 keys
|
||||
found_files = {} # Mapping of original file info to actual S3 keys
|
||||
for norm_filename, file_info in normalized_files_mapping.items():
|
||||
norm_key = normalize_string(norm_filename.replace('.txt', '')) # Normalize without '.txt'
|
||||
actual_key = normalized_key_mapping.get(norm_key)
|
||||
if actual_key:
|
||||
found_files[file_info['File Name']] = actual_key
|
||||
else:
|
||||
# File not found; will be handled later
|
||||
pass
|
||||
|
||||
# Display the results
|
||||
total_found = len(found_files)
|
||||
print(f"\nTotal files found: {total_found}")
|
||||
for fname, key in found_files.items():
|
||||
print(f"Found: {key}")
|
||||
|
||||
# Identify files not found
|
||||
found_filenames = set(found_files.keys())
|
||||
not_found_files = [file for file in files if file['File Name'] not in found_filenames]
|
||||
|
||||
# Write not found files to CSV
|
||||
if not_found_files:
|
||||
write_not_found_csv(not_found_files, NOT_FOUND_CSV)
|
||||
else:
|
||||
print("All files were found. No entries to write to not_found CSV.")
|
||||
|
||||
# ----------------- Copy Functionality -----------------
|
||||
|
||||
print("\nStarting to copy files to the destination prefix...")
|
||||
|
||||
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as copy_executor:
|
||||
copy_futures = []
|
||||
for filename, source_key in found_files.items():
|
||||
destination_key = os.path.join(DESTINATION_PREFIX, os.path.basename(source_key))
|
||||
copy_futures.append(
|
||||
copy_executor.submit(copy_file, s3_client, BUCKET_NAME, source_key, destination_key)
|
||||
)
|
||||
|
||||
# Optionally, wait for all copy operations to complete
|
||||
for future in as_completed(copy_futures):
|
||||
try:
|
||||
future.result()
|
||||
except Exception as exc:
|
||||
print(f"Copy operation generated an exception: {exc}")
|
||||
|
||||
print("File copying completed.")
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,197 @@
|
||||
import boto3
|
||||
import pandas as pd
|
||||
from io import StringIO
|
||||
import os
|
||||
import json
|
||||
import requests
|
||||
|
||||
|
||||
"""
|
||||
This script is used to process files from a source S3 bucket and trigger the pipeline for processing.
|
||||
|
||||
The script does the following:
|
||||
1. Load files from the source S3 bucket based on the directory prefix
|
||||
2. Create a batch in the client bucket
|
||||
3. Copy files to the client bucket and add tags
|
||||
4. Move the original files to a processed folder in the source bucket
|
||||
5. Create a contract list and send it to the pipeline API
|
||||
6. Update the CSV with the processed batch details
|
||||
7. Trigger the pipeline API
|
||||
|
||||
The script uses the following variables:
|
||||
- source_bucket: The source S3 bucket name
|
||||
(Client bucket where all source files are located.
|
||||
NOTE: This function will move the files to processed folder in the source bucket, so the source prefix will be empty after the execution)
|
||||
|
||||
- api_trigger_url: The URL of the pipeline trigger API ( This explicit url can be reaplced by using AWS SDK to get the API URL)
|
||||
|
||||
- create_batch_url: The URL of the create batch API
|
||||
|
||||
- csv_file: The name of the CSV file to store processed batch details
|
||||
(Usually we add relevant bucket of files to track this, e.g. Facility_files.csv )
|
||||
|
||||
- processed_csv_s3_key: The S3 key to store the processed CSV file
|
||||
|
||||
- client_bucket: The client S3 bucket name
|
||||
|
||||
- directory_prefix_list: A list of directory prefixes to search for files in the source bucket
|
||||
|
||||
- client_name: The name of the client (for logs)
|
||||
|
||||
- username: The username of the user triggering the pipeline (for logs)
|
||||
|
||||
The script can be scheduled to run periodically to process files from the source bucket.
|
||||
|
||||
Add this code to a lambda function, ensure there is a lambda layer for pandas
|
||||
AWS provides a layer for pandas, you can use that or create your own layer (AWSSDKPandas-Python310)
|
||||
|
||||
Once the function is created, we can create an event trigger for the lambda function to run at a specific time or a a specific interval.
|
||||
|
||||
Note: The script assumes that the pipeline trigger API and create batch API are already set up and working correctly.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
|
||||
source_bucket = 'centene-texas-files'
|
||||
api_trigger_url = 'https://<APISTAGE>.execute-api.us-east-2.amazonaws.com/dev/trigger-pipeline'
|
||||
create_batch_url = "https://<APISTAGE>.execute-api.us-east-2.amazonaws.com/dev/create-batch"
|
||||
csv_file = 'ancillary_files.csv' #file count need to as prefix
|
||||
processed_csv_s3_key = f'processed_batch_ids/{csv_file}'
|
||||
client_bucket = 'doczyai-use2-u-cn1-s3-textract-processing-001'
|
||||
client_name = 'Centene-texas'
|
||||
username = '<Username>'
|
||||
|
||||
directory_prefix_list = [
|
||||
'for-textract/'
|
||||
# Can add more prefixes here if needed
|
||||
|
||||
|
||||
# 'TX-Legacy-agreements_2/batch2/for-processing/',
|
||||
# 'TX-Legacy-agreements_2/batch2/for-processing/',
|
||||
# 'TX-Legacy-agreements_3/batch3/for-processing/',
|
||||
# 'TX-Legacy-agreements_4/batch4/for-processing/',
|
||||
# 'TX-Legacy-Amendments/TX-Legacy-Amendments/for-processing/'
|
||||
]
|
||||
|
||||
|
||||
def lambda_handler(event, context):
|
||||
s3_resource = boto3.resource('s3')
|
||||
|
||||
for directory_prefix in directory_prefix_list:
|
||||
bucket_obj = s3_resource.Bucket(source_bucket)
|
||||
files = [obj.key for obj in bucket_obj.objects.filter(Prefix=directory_prefix) if obj.key.endswith('.Pdf') or obj.key.endswith('.pdf')]
|
||||
|
||||
if files:
|
||||
print(f"Processing files for directory prefix: {directory_prefix}")
|
||||
process_files(source_bucket, directory_prefix, files[:60], create_batch_url, api_trigger_url)
|
||||
break
|
||||
else:
|
||||
print(f"No files found for directory prefix: {directory_prefix}. Moving to the next prefix...")
|
||||
|
||||
def process_files(source_bucket, directory_prefix, files, create_batch_url, api_trigger_url):
|
||||
s3_client = boto3.client('s3')
|
||||
|
||||
|
||||
# Load existing CSV from S3 if it exists
|
||||
existing_df = None
|
||||
try:
|
||||
csv_obj = s3_client.get_object(Bucket=source_bucket, Key=processed_csv_s3_key)
|
||||
existing_df = pd.read_csv(csv_obj['Body'])
|
||||
print(f"Loaded existing CSV with {len(existing_df)} records")
|
||||
except s3_client.exceptions.NoSuchKey:
|
||||
print("No existing CSV found. A new one will be created.")
|
||||
|
||||
batch_id, landing_zone = create_batch(client_bucket, create_batch_url)
|
||||
if batch_id == 'failed_cases':
|
||||
print("Batch creation failed. Skipping this batch...")
|
||||
return
|
||||
|
||||
contract_list = []
|
||||
for s3_key in files:
|
||||
|
||||
filename = os.path.basename(s3_key).replace('.Pdf', '.pdf')
|
||||
new_s3_key = f'contracts-landing-zone/{batch_id}/{filename}'
|
||||
|
||||
# Copy the file to the new key with .pdf extension
|
||||
copy_source = {'Bucket': source_bucket, 'Key': s3_key}
|
||||
s3_client.copy_object(CopySource=copy_source, Bucket=client_bucket, Key=new_s3_key)
|
||||
|
||||
# Add tags to the copied file
|
||||
s3_client.put_object_tagging(
|
||||
Bucket=client_bucket,
|
||||
Key=new_s3_key,
|
||||
Tagging={
|
||||
'TagSet': [
|
||||
{
|
||||
'Key': 'BatchId',
|
||||
'Value': batch_id
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
# Move the original file to the processed subfolder
|
||||
processed_s3_key = f'processed/{os.path.basename(s3_key)}'
|
||||
s3_client.copy_object(CopySource=copy_source, Bucket=source_bucket, Key=processed_s3_key)
|
||||
s3_client.delete_object(Bucket=source_bucket, Key=s3_key)
|
||||
|
||||
# Add file details to contract list
|
||||
contract_list.append({
|
||||
"contract_name": filename,
|
||||
"groups": ["A", "C"], # This should be dynamic based on your requirements
|
||||
"contract_source_path": new_s3_key
|
||||
})
|
||||
print(f"Uploaded {len(files)} files to S3 bucket for batch id {batch_id}")
|
||||
data = {
|
||||
"s3_bucket": client_bucket,
|
||||
"batch_id": batch_id,
|
||||
"client_name": client_name,
|
||||
"username": username,
|
||||
"contract_list": contract_list
|
||||
}
|
||||
|
||||
# Convert contract list to DataFrame and append to existing DataFrame
|
||||
new_df = pd.DataFrame(contract_list)
|
||||
new_df['batch_id'] = batch_id
|
||||
if existing_df is not None:
|
||||
existing_df = pd.concat([existing_df, new_df])
|
||||
else:
|
||||
existing_df = new_df
|
||||
|
||||
# Upload updated CSV back to S3
|
||||
csv_buf = StringIO()
|
||||
existing_df.to_csv(csv_buf, header=True, index=False)
|
||||
csv_buf.seek(0)
|
||||
|
||||
s3_client.put_object(Bucket=source_bucket, Body=csv_buf.getvalue(), Key=processed_csv_s3_key)
|
||||
|
||||
print(f"Saved updated batch details to {processed_csv_s3_key} and sending to API...")
|
||||
response = requests.post(api_trigger_url, json=data)
|
||||
print(response.text)
|
||||
print(response.status_code)
|
||||
|
||||
return {
|
||||
"statusCode": 200,
|
||||
"body": json.dumps("Processed files successfully")
|
||||
}
|
||||
|
||||
def create_batch(client_bucket, create_batch_url):
|
||||
myobj = { "client-bucket-name": client_bucket }
|
||||
response = requests.post(create_batch_url, json=myobj)
|
||||
if response.status_code >= 200 and response.status_code < 300:
|
||||
try:
|
||||
response_body = json.loads(json.loads(response.text)['body'])
|
||||
new_batch_id = response_body['batch_id']
|
||||
landing_zone = response_body['landing_zone']
|
||||
except:
|
||||
print(myobj)
|
||||
print(response.text)
|
||||
new_batch_id = 'failed_cases'
|
||||
landing_zone = 'contracts_landing_zone'
|
||||
else:
|
||||
print(response.text)
|
||||
new_batch_id = 'failed_cases'
|
||||
landing_zone = 'contracts_landing_zone'
|
||||
|
||||
return new_batch_id, landing_zone
|
||||
@@ -0,0 +1,181 @@
|
||||
import boto3
|
||||
import os
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
import shutil
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
|
||||
"""
|
||||
|
||||
This script is used to move files from one S3 bucket to another S3 bucket based on the batch_ids provided in the CSV file.
|
||||
The script will move the text files, stuck files, PDF files and invalid files to the destination bucket.
|
||||
The copy operation is done using multithreading to speed up the process. Use the max_workers parameter to adjust the number of threads.
|
||||
|
||||
This script is NOT CLIENT SPECIFIC. It is a generic script that can be used for any client.
|
||||
|
||||
Usage:
|
||||
1. Update the configuration section with the required parameters.
|
||||
2. Run the script.
|
||||
|
||||
Note:
|
||||
- Ensure you have the necessary permissions to read from the source bucket and write to the destination bucket (Admin role is preferred).
|
||||
|
||||
"""
|
||||
|
||||
def log_error_to_file(key, error_message, error_log_file):
|
||||
# Log error to file
|
||||
# Create new file if it doesnt exist
|
||||
if not os.path.exists(error_log_file):
|
||||
with open(error_log_file, 'w') as f:
|
||||
f.write("Error Log\n")
|
||||
f.write("---------\n")
|
||||
|
||||
with open(error_log_file, 'a') as f:
|
||||
f.write(f"Error with file: {key} - {error_message}\n")
|
||||
|
||||
def list_filtered_files(bucket_name, prefix, start_date, end_date, profile_name):
|
||||
start_date = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
|
||||
end_date = datetime.fromisoformat(end_date.replace('Z', '+00:00'))
|
||||
|
||||
session = boto3.Session(profile_name=profile_name)
|
||||
s3_client = session.client('s3')
|
||||
|
||||
paginator = s3_client.get_paginator('list_objects_v2')
|
||||
page_iterator = paginator.paginate(Bucket=bucket_name, Prefix=prefix)
|
||||
|
||||
filtered_files = []
|
||||
for page in page_iterator:
|
||||
if 'Contents' in page:
|
||||
for obj in page['Contents']:
|
||||
last_modified = obj['LastModified']
|
||||
if start_date <= last_modified <= end_date:
|
||||
filtered_files.append(obj['Key'])
|
||||
|
||||
return filtered_files
|
||||
|
||||
def copy_single_file(s3_client, source_bucket, destination_bucket, key, destination_prefix, error_log_file):
|
||||
try:
|
||||
copy_source = {'Bucket': source_bucket, 'Key': key}
|
||||
destination_key = os.path.join(destination_prefix, os.path.basename(key))
|
||||
print(f"Copying {key} to {destination_key}")
|
||||
s3_client.copy(copy_source, destination_bucket, destination_key)
|
||||
except Exception as e:
|
||||
error_message = str(e)
|
||||
print(f"Error copying file {key}: {error_message}")
|
||||
# Log error to file
|
||||
log_error_to_file(key, error_message, error_log_file)
|
||||
|
||||
def copy_files_to_destination(source_bucket, destination_bucket, file_keys, profile_name, destination_prefix, max_workers, error_log_file):
|
||||
session = boto3.Session(profile_name=profile_name)
|
||||
s3_client = session.client('s3')
|
||||
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
futures = [executor.submit(copy_single_file, s3_client, source_bucket, destination_bucket, key, destination_prefix, error_log_file) for key in file_keys]
|
||||
|
||||
for future in as_completed(futures):
|
||||
try:
|
||||
future.result()
|
||||
except Exception as e:
|
||||
print(f"Error in thread: {e}")
|
||||
|
||||
print("Copy completed.")
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
|
||||
# ------------------------------ Configuration ------------------------------
|
||||
|
||||
|
||||
# Error log file path
|
||||
# There is usually no errors in this script, but in case of any errors, they will be logged to this file
|
||||
error_log_file = "error_log_batch5.txt"
|
||||
|
||||
|
||||
# Main script
|
||||
# Fetch the batch_ids csv file from the client bucket ('centene-national-contracting-files')/processed_batch_ids/ folder
|
||||
csv_file_path = "batch5_diff_files.csv"
|
||||
|
||||
# Source and destination bucket names. Where source bucket is the textract pipeline where all files reside in their batch folders
|
||||
source_bucket = 'doczyai-use2-u-cn1-s3-textract-processing-001'
|
||||
|
||||
# Destination bucket is the client bucket where we move the files to satge it for DS team
|
||||
destination_bucket = 'centene-national-contracting-files'
|
||||
|
||||
# Depending on what batch files we want to move, we can change the destination prefix
|
||||
destination_prefix = 'batch_5_priority_files/txt_files/'
|
||||
|
||||
# If the batch size in scheduler is large, there is a possibility of throttling and files getting stuck in the textract pipeline
|
||||
# We move these stuck files to a separate folder so that it can be sent for reprocessing
|
||||
stuck_destination_prefix = 'batch_5_priority_files/throttled_files_final/'
|
||||
|
||||
# PDF files are required for the MCS team to review the contracts
|
||||
pdf_files_destination = f'batch_5_priority_files/pdf_files/'
|
||||
|
||||
# Less than 1% of the files are invalid files. These files are moved to a separate folder
|
||||
invalid_files_destination = f'batch_5_priority_files/invalid_files/'
|
||||
|
||||
# Date range for files to be moved so that we can limit the list files in bucket operation
|
||||
start_date = '2024-07-07T00:00:00Z' # ISO 8601 format
|
||||
end_date = '2024-12-31T23:59:59Z' # ISO 8601 format
|
||||
|
||||
# If you are using AWS CLI profiles, provide the profile name here to authenticate
|
||||
profile_name = 'doczy_uat'
|
||||
max_workers = 50 # Adjust the number of threads as needed
|
||||
|
||||
|
||||
# ------------------------------ Script ------------------------------
|
||||
|
||||
batch_ids_df = pd.read_csv(csv_file_path)
|
||||
batch_ids = batch_ids_df['batch_id'].unique()
|
||||
|
||||
print(f"Total number of batches: {len(batch_ids)}")
|
||||
print(f"Batch IDs: {batch_ids}")
|
||||
|
||||
text_count = 0
|
||||
stuck_count = 0
|
||||
pdf_count = 0
|
||||
invalid_count = 0
|
||||
|
||||
# Bulk copy files from S3 for all batch_ids
|
||||
for batch_id in batch_ids:
|
||||
prefix = f'contract-text-file/{batch_id}/'
|
||||
stuck_files_path = f'textract-sender-staging-pdfs/{batch_id}/'
|
||||
pdf_files_path = f'textract-receiver-processed-pdfs/{batch_id}'
|
||||
invalid_files_path = f'invalid-pdfs/{batch_id}'
|
||||
|
||||
# List filtered text files
|
||||
filtered_files = list_filtered_files(source_bucket, prefix, start_date, end_date, profile_name)
|
||||
print(f"Batch_id: {batch_id} Filtered files: {len(filtered_files)}")
|
||||
text_count += len(filtered_files)
|
||||
|
||||
# Throttled files list
|
||||
stuck_files = list_filtered_files(source_bucket, stuck_files_path, start_date, end_date, profile_name)
|
||||
print(f"Batch_id: {batch_id} Stuck files: {len(stuck_files)}")
|
||||
stuck_count += len(stuck_files)
|
||||
|
||||
# PDF file list
|
||||
pdf_files = list_filtered_files(source_bucket, pdf_files_path, start_date, end_date, profile_name)
|
||||
print(f"Batch_id: {batch_id} PDF files: {len(pdf_files)}")
|
||||
pdf_count += len(pdf_files)
|
||||
|
||||
# Invalid files list
|
||||
invalid_files = list_filtered_files(source_bucket, invalid_files_path, start_date, end_date, profile_name)
|
||||
print(f"Batch_id: {batch_id} Invalid files: {len(invalid_files)}")
|
||||
invalid_count += len(invalid_files)
|
||||
|
||||
# Copy files to destination using multithreading
|
||||
copy_files_to_destination(source_bucket, destination_bucket, filtered_files, profile_name, destination_prefix, max_workers, error_log_file)
|
||||
copy_files_to_destination(source_bucket, destination_bucket, stuck_files, profile_name, stuck_destination_prefix, max_workers, error_log_file)
|
||||
copy_files_to_destination(source_bucket, destination_bucket, pdf_files, profile_name, pdf_files_destination, max_workers, error_log_file)
|
||||
copy_files_to_destination(source_bucket, destination_bucket, invalid_files, profile_name, invalid_files_destination, max_workers, error_log_file)
|
||||
|
||||
print(f"Total text files copied: {text_count}")
|
||||
print(f"Total stuck files copied: {stuck_count}")
|
||||
print(f"Total pdf files copied: {pdf_count}")
|
||||
print(f"Total invalid files copied: {invalid_count}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user