Merged in feature/ops_scripts (pull request #333)
Feature/ops scripts * Added comments to Aryan's script and added some more scripts * Search and Copy Script uploaded as a Python Notebook - with comments and markdown * Merged main into feature/ops_scripts * Merged main into feature/ops_scripts Approved-by: Michael McGuinness Approved-by: Chris Stobie
This commit is contained in:
Umang Shailesh Mistry
committed by
Michael McGuinness
Michael McGuinness
parent
91d3680ee4
commit
c210052952
@@ -3,6 +3,18 @@ import csv
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
"""
|
||||
This script is used for CNC diff analysis. It compares two folders (source1 and source2) and writes the comparison results to a CSV file.
|
||||
|
||||
source1_type and source2_type should be one of the following:
|
||||
- local: for local directories
|
||||
- s3: for S3 folders
|
||||
- csv: for CSV files
|
||||
|
||||
S3 credentials should be stored in the AWS credentials file under the profile name 'temp_cred' or can be changed in the aws credentials set.
|
||||
|
||||
"""
|
||||
|
||||
def list_s3_files(bucket_name, folder):
|
||||
s3 = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
paginator = s3.get_paginator('list_objects_v2')
|
||||
|
||||
@@ -4,6 +4,20 @@ import glob
|
||||
from PyPDF2 import PdfReader
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
"""
|
||||
This script will collect the following information for all PDF files in a directory and its subdirectories:
|
||||
- File Name
|
||||
- File Path
|
||||
- Page Count
|
||||
|
||||
The information will be saved in a CSV file.
|
||||
|
||||
NOTE: This script is required mainly in the pre-doczy analysis. There might not be need for this since CNC project is already processed through the textract pipeline.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
|
||||
def get_pdf_info(pdf_path):
|
||||
try:
|
||||
with open(pdf_path, 'rb') as file:
|
||||
|
||||
@@ -4,6 +4,17 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from collections import defaultdict
|
||||
import csv
|
||||
|
||||
"""
|
||||
This script searches for files in a given directory based on a list of filenames in a CSV file.
|
||||
The CSV file should have a column named 'filename' containing the filenames to search for.
|
||||
|
||||
This is maily used to find files that are not present in the s3 or cannot be located.
|
||||
In that case, we use the list of missing files (in the CSV) to search for them in the T drive.
|
||||
|
||||
This script may not be needed as all CNC batches have been staged for execution.
|
||||
|
||||
"""
|
||||
|
||||
def build_file_cache(base_directory):
|
||||
file_cache = defaultdict(list)
|
||||
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"This is an adhoc python notebook that mainly does some dups analysis, file movements and debugging. \n",
|
||||
"This is not something that is usually repeated and is pushed only for reference. "
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
|
||||
@@ -1,6 +1,19 @@
|
||||
import boto3
|
||||
import concurrent.futures
|
||||
|
||||
|
||||
"""
|
||||
This script moves files with a specific extension from one folder to another in an S3 bucket.
|
||||
It uses multiple threads to move files in parallel.
|
||||
|
||||
The script uses a temporary AWS profile called 'temp_cred' to authenticate with AWS.
|
||||
The profile should be configured in the ~/.aws/credentials file.
|
||||
|
||||
Mainly used to copy and delete (move) files in the same bucket. Typically used to move redundant dups back to the batch TXT folder.
|
||||
Can be used to copy .txt / .pdf files from one folder to another in the same bucket.
|
||||
|
||||
"""
|
||||
|
||||
s3 = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def move_file(bucket_name, src_folder, dest_folder, file_key):
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Setting up imports that we will need, particularly - csv, boto3, ThreadPoolExecutor and pandas"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 36,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import csv\n",
|
||||
"import boto3\n",
|
||||
"from concurrent.futures import ThreadPoolExecutor\n",
|
||||
"from botocore.exceptions import ClientError\n",
|
||||
"import pandas as pd\n",
|
||||
"\n",
|
||||
"s3 = boto3.Session(profile_name='temp_cred').client('s3')"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"This function use pagination to iterate through all objects in the specified s3 bucket and builds a file cache with all s3 keys. This is meant to be run only-once and is expected to take 8-10 mins to complete"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 37,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"def list_all_files(bucket_name):\n",
|
||||
" all_keys = []\n",
|
||||
" try:\n",
|
||||
" paginator = s3.get_paginator('list_objects_v2')\n",
|
||||
" for page in paginator.paginate(Bucket=bucket_name):\n",
|
||||
" if 'Contents' in page:\n",
|
||||
" all_keys.extend([item['Key'] for item in page['Contents']])\n",
|
||||
" except ClientError as e:\n",
|
||||
" print(f\"Error listing files in bucket {bucket_name}: {e}\")\n",
|
||||
" return all_keys\n",
|
||||
"\n",
|
||||
"bucket_name = 'centene-national-contracting-files'\n",
|
||||
"all_keys = list_all_files(bucket_name)"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Searching v1 - using complete matching of the filenames in the s3 bucket. If ```copy``` is set to ```True```, then the files found will also be copied to the specified prefix."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 44,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"CSV processing completed.\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"# Complete matching\n",
|
||||
"def check_file_in_bucket(filename, all_keys):\n",
|
||||
" locations = [key for key in all_keys if (key.lower().endswith(filename.lower()))]\n",
|
||||
" if copy:\n",
|
||||
" for l in locations:\n",
|
||||
" s3.copy_object(\n",
|
||||
" CopySource={'Bucket': bucket_name, 'Key': l},\n",
|
||||
" Bucket=bucket_name,\n",
|
||||
" Key=destination_prefix + \"/\" + filename\n",
|
||||
" )\n",
|
||||
" exists = bool(locations)\n",
|
||||
" return filename, exists, locations\n",
|
||||
"\n",
|
||||
"def process_csv(input_csv, output_csv):\n",
|
||||
" \n",
|
||||
" with open(input_csv, mode='r', newline='', encoding='utf8') as infile, open(output_csv, mode='w', newline='', encoding='utf8') as outfile:\n",
|
||||
" reader = csv.reader(infile)\n",
|
||||
" writer = csv.writer(outfile)\n",
|
||||
" writer.writerow(['File Name', 'exists', 'locations']) # Write header\n",
|
||||
"\n",
|
||||
" with ThreadPoolExecutor(max_workers = 50) as executor:\n",
|
||||
" futures = [executor.submit(check_file_in_bucket, row[0][:-4] + file_extension, all_keys) for row in reader]\n",
|
||||
" \n",
|
||||
" for future in futures:\n",
|
||||
" filename, exists, locations = future.result()\n",
|
||||
" writer.writerow([filename, exists, locations])\n",
|
||||
"\n",
|
||||
"# Input csv with all file names listed without extension in the first column\n",
|
||||
"input_csv = '../CNC/batch4/batch4_missing_files_2.csv'\n",
|
||||
"# Path to output that will be generated with the following columns = File Name, exists, locations\n",
|
||||
"output_csv = '../CNC/batch4/batch4_missing_files_2_paths_txt.csv'\n",
|
||||
"# If copy = True, the files found will be copied over to this destination in s3 (User Keys needs copy permission)\n",
|
||||
"destination_prefix = 'batch_4_priority_files/hotfix_diff_121124/TXT_FILES'\n",
|
||||
"copy = True\n",
|
||||
"# File Extension that will be added to the end of the file names - to look for\n",
|
||||
"file_extension = '.txt'\n",
|
||||
"\n",
|
||||
"process_csv(input_csv, output_csv)\n",
|
||||
"print(\"CSV processing completed.\")"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"Searching v2 - using partial matching of the filenames in the s3 bucket. In this specific case, we check the first 10 and last 10 characters of the filename (This logic can be changed in the 3rd line, as required). If ```copy``` is set to ```True```, then the files found will also be copied to the specified prefix."
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"# Partial matching\n",
|
||||
"def check_file_in_bucket(filename, all_keys):\n",
|
||||
" locations = [key for key in all_keys if (filename[:10].lower() in key.lower() and key.lower().endswith(filename[-10:].lower()))]\n",
|
||||
" if copy:\n",
|
||||
" for l in locations:\n",
|
||||
" s3.copy_object(\n",
|
||||
" CopySource={'Bucket': bucket_name, 'Key': l},\n",
|
||||
" Bucket=bucket_name,\n",
|
||||
" Key=destination_prefix + \"/\" + filename\n",
|
||||
" )\n",
|
||||
" exists = bool(locations)\n",
|
||||
" return filename, exists, locations\n",
|
||||
"\n",
|
||||
"def process_csv(input_csv, output_csv):\n",
|
||||
" \n",
|
||||
" with open(input_csv, mode='r', newline='', encoding='utf8') as infile, open(output_csv, mode='w', newline='', encoding='utf8') as outfile:\n",
|
||||
" reader = csv.reader(infile)\n",
|
||||
" writer = csv.writer(outfile)\n",
|
||||
" writer.writerow(['File Name', 'exists', 'locations']) # Write header\n",
|
||||
"\n",
|
||||
" with ThreadPoolExecutor(max_workers = 50) as executor:\n",
|
||||
" futures = [executor.submit(check_file_in_bucket, row[0][:-4] + file_extension, all_keys) for row in reader]\n",
|
||||
" \n",
|
||||
" for future in futures:\n",
|
||||
" filename, exists, locations = future.result()\n",
|
||||
" writer.writerow([filename, exists, locations])\n",
|
||||
"\n",
|
||||
"# Input csv with all file names listed without extension in the first column\n",
|
||||
"input_csv = '../CNC/batch4/batch4_missing_files_2.csv'\n",
|
||||
"# Path to output that will be generated with the following columns = File Name, exists, locations\n",
|
||||
"output_csv = '../CNC/batch4/batch4_missing_files_2_paths_txt.csv'\n",
|
||||
"# If copy = True, the files found will be copied over to this destination in s3 (User Keys needs copy permission)\n",
|
||||
"destination_prefix = 'batch_4_priority_files/hotfix_diff_121124/TXT_FILES'\n",
|
||||
"copy = True\n",
|
||||
"# File Extension that will be added to the end of the file names - to look for\n",
|
||||
"file_extension = '.txt'\n",
|
||||
"\n",
|
||||
"process_csv(input_csv, output_csv)\n",
|
||||
"print(\"CSV processing completed.\")"
|
||||
]
|
||||
}
|
||||
],
|
||||
"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
|
||||
}
|
||||
@@ -3,6 +3,16 @@ import boto3
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
"""
|
||||
This script is used to search for files in an S3 bucket and copy them to a new location.
|
||||
This is used when CNC team comes back with a list of missing files and we need to search for them in the S3 bucket.
|
||||
|
||||
We usually run this script to first just search for the files without copying, then we analyze the results and if needed, we run the script again with copy=True.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
|
||||
s3 = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def list_all_files(bucket_name):
|
||||
|
||||
@@ -2,6 +2,16 @@ import boto3
|
||||
import pandas as pd
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
"""
|
||||
This script is used to search for files in a specific S3 bucket with a specific prefix and copy them to another S3 bucket with a different prefix.
|
||||
|
||||
This is similar to the script s3_search_and_copy.py, but this script is designed to be used with a CSV file that contains a list of file names to search for in the prefix to limit the scope.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
|
||||
|
||||
s3_client = boto3.Session(profile_name='temp_cred').client('s3')
|
||||
|
||||
def get_filenames_from_s3(bucket, prefix):
|
||||
|
||||
@@ -3,6 +3,16 @@ import re
|
||||
import csv
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
"""
|
||||
This script reads all the text files in a specific S3 bucket and prefix, and extracts the last page number from each file.
|
||||
The output is written to a CSV file with the filename and the last page number.
|
||||
|
||||
This script is not used often and was needed for an adhoc analysis. It is not part of the regular data pipeline.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
|
||||
BUCKET_NAME = "centene-national-contracting-files"
|
||||
PREFIX = "batch_1_priority_files/text_files/"
|
||||
OUTPUT_CSV = "batch1/batch1_s3_files_page_counts.csv"
|
||||
|
||||
@@ -13,7 +13,7 @@ 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.
|
||||
4. **Export Data**: Export the cost data and bucket record counts to CSV files. (This is currently disabled by default.)
|
||||
|
||||
"""
|
||||
|
||||
@@ -82,15 +82,13 @@ def process_bucket(bucket_name, start_date_obj, end_date_obj, profile_name):
|
||||
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
|
||||
end_date = '2024-11-30' # Replace with your end date
|
||||
client_buckets = ['highmark-files', 'la-care-files',
|
||||
'centene-fidelis-files',
|
||||
'centene-healthnet-files',
|
||||
@@ -148,6 +146,29 @@ def main():
|
||||
total_costs = sum([cost['Cost'] for cost in cost_data])
|
||||
cost_data.append({'Category': 'Total', 'Item': 'Total', 'Cost': total_costs})
|
||||
|
||||
# Adding up all costs that are not Amazon Textract and other costs as bedrock costs
|
||||
# then based on the costs, we split the support and tax costs respective to the textract vs bedrock costs
|
||||
|
||||
textract_costs = service_costs.get('Amazon Textract', 0.0)
|
||||
bedrock_costs = total_costs - textract_costs
|
||||
|
||||
support_costs_textract = support_costs.get('Support', 0.0) * (textract_costs / total_costs)
|
||||
support_costs_bedrock = support_costs.get('Support', 0.0) * (bedrock_costs / total_costs)
|
||||
|
||||
tax_costs_textract = tax_costs.get('Tax', 0.0) * (textract_costs / total_costs)
|
||||
tax_costs_bedrock = tax_costs.get('Tax', 0.0) * (bedrock_costs / total_costs)
|
||||
|
||||
print(f"Pro-rated Support + Tax costs for Textract based on total costs: ${support_costs_textract + tax_costs_textract:.2f}")
|
||||
print(f"Pro-rated Support + Tax costs for Bedrock based on total costs: ${support_costs_bedrock + tax_costs_bedrock:.2f}")
|
||||
|
||||
|
||||
total_textract_costs = textract_costs + support_costs_textract + tax_costs_textract
|
||||
total_bedrock_costs = bedrock_costs + support_costs_bedrock + tax_costs_bedrock
|
||||
|
||||
print("\nAWS Service Costs pro-rated taxes by service :")
|
||||
print(f" TOTAL Amazon Textract: ${total_textract_costs:.2f}")
|
||||
print(f" TOTAL Amazon Bedrock: ${total_bedrock_costs:.2f}")
|
||||
|
||||
# Process each bucket and count records
|
||||
print("\nBucket-wise Record Counts:")
|
||||
bucket_data = []
|
||||
@@ -156,18 +177,39 @@ def main():
|
||||
print(f" {bucket}: {total_records} records")
|
||||
bucket_data.append({'Bucket': bucket, 'Records': total_records})
|
||||
|
||||
# Combine all record counts for buckets and calcuate the percent of total for each bucket. Ingoring buckets with 0 records
|
||||
|
||||
total_records = sum([record['Records'] for record in bucket_data])
|
||||
for record in bucket_data:
|
||||
if record['Records'] > 0:
|
||||
record['Percent of Total'] = (record['Records'] / total_records) * 100
|
||||
|
||||
|
||||
# Printing the record weightage for each bucket
|
||||
print("\nBucket-wise Record Weightage:")
|
||||
|
||||
for record in bucket_data:
|
||||
if record['Records'] > 0:
|
||||
print(f" {record['Bucket']}: {record['Percent of Total']:.2f}%")
|
||||
# Calcualting the textract cost with respect to the percent of total
|
||||
textract_cost_bucket = total_textract_costs * record['Percent of Total'] / 100
|
||||
print(f" {record['Bucket']}: ${textract_cost_bucket:.2f}")
|
||||
|
||||
print(f'Total textract documents from buckets: {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'.")
|
||||
# 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'.")
|
||||
# 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,283 @@
|
||||
{
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"metadata": {},
|
||||
"source": [
|
||||
"This is an adhoc script that is used to ease the process of archiving. \n",
|
||||
"Initially there was a scirpt that would run the process of archiving but there were edge cases where that script would fail due to the file names having nuances. \n",
|
||||
"\n",
|
||||
"This script simplifies that process. \n",
|
||||
"It will take input of S3 URIs of folders that need to be archived, \n",
|
||||
"e.g. \n",
|
||||
"s3://centene-national-contracting-files/batch_3_priority_files/3b_missing_files_103024/\n",
|
||||
"\n",
|
||||
"And move it to \n",
|
||||
"\n",
|
||||
"s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/3b_missing_files_103024/\n",
|
||||
"\n",
|
||||
"This archived folders within the batch folder. \n",
|
||||
"\n",
|
||||
"There is an option for dryrun which can be set as true which only outputs the \"aws s3 mv --recursive\" command. \n",
|
||||
" This can be used to run it manually to have more control over the folders that get archived \n",
|
||||
"\n",
|
||||
"Or we can set dry_run as False and let the subprocess run the aws cli command.\n",
|
||||
"\n",
|
||||
"This is setup as ipynb to ensure some more control over variables and paths and iterating accordingly "
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 6,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": [
|
||||
"import subprocess\n",
|
||||
"import logging\n",
|
||||
"import posixpath\n",
|
||||
"import sys\n",
|
||||
"\n",
|
||||
"# ----------------------------- Configuration ----------------------------- #\n",
|
||||
"\n",
|
||||
"# List of S3 URIs to archive\n",
|
||||
"S3_URIS = [\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/3b_missing_files_103024/',\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/dups_rerun_outputs/',\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/dups-for-textract/',\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/for-textract/',\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/hotfix_rerun_112724/',\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/redundant_dups/',\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/tables-for-textract/'\n",
|
||||
" 's3://centene-national-contracting-files/batch_3_priority_files/throttled_files/'\n",
|
||||
" # Add more S3 URIs as needed\n",
|
||||
"]\n",
|
||||
"\n",
|
||||
"# AWS Profile to use\n",
|
||||
"AWS_PROFILE = 'doczy_uat' # Replace with your AWS profile name\n",
|
||||
"\n",
|
||||
"# Dry Run Configuration\n",
|
||||
"DRY_RUN = False # Set to False to execute the commands\n",
|
||||
"\n",
|
||||
"# Logging Configuration\n",
|
||||
"LOG_FILE = 's3_mv_commands.log' # Log file name\n",
|
||||
"\n",
|
||||
"# ---------------------------- End Configuration --------------------------- #\n",
|
||||
"\n",
|
||||
"# Initialize logging\n",
|
||||
"logging.basicConfig(\n",
|
||||
" filename=LOG_FILE,\n",
|
||||
" filemode='a',\n",
|
||||
" format='%(asctime)s - %(levelname)s - %(message)s',\n",
|
||||
" level=logging.INFO\n",
|
||||
")\n",
|
||||
"\n",
|
||||
"# Initialize a thread-safe print function\n",
|
||||
"from threading import Lock\n",
|
||||
"print_lock = Lock()\n",
|
||||
"\n",
|
||||
"def log(message, level='info'):\n",
|
||||
" \"\"\"\n",
|
||||
" Logs messages to both the console and a log file.\n",
|
||||
"\n",
|
||||
" :param message: The message to log.\n",
|
||||
" :param level: The logging level ('info', 'error', 'warning', 'debug').\n",
|
||||
" \"\"\"\n",
|
||||
" with print_lock:\n",
|
||||
" if level == 'info':\n",
|
||||
" logging.info(message)\n",
|
||||
" print(message)\n",
|
||||
" elif level == 'error':\n",
|
||||
" logging.error(message)\n",
|
||||
" print(f\"ERROR: {message}\")\n",
|
||||
" elif level == 'warning':\n",
|
||||
" logging.warning(message)\n",
|
||||
" print(f\"WARNING: {message}\")\n",
|
||||
" elif level == 'debug':\n",
|
||||
" logging.debug(message)\n",
|
||||
" print(f\"DEBUG: {message}\")\n",
|
||||
"\n",
|
||||
"def parse_s3_uri(s3_uri):\n",
|
||||
" \"\"\"\n",
|
||||
" Parses an S3 URI and returns the bucket and key.\n",
|
||||
"\n",
|
||||
" :param s3_uri: The S3 URI (e.g., s3://bucket/key)\n",
|
||||
" :return: Tuple of (bucket, key)\n",
|
||||
" \"\"\"\n",
|
||||
" if not s3_uri.startswith('s3://'):\n",
|
||||
" raise ValueError(f\"Invalid S3 URI: {s3_uri}\")\n",
|
||||
" parts = s3_uri[5:].split('/', 1)\n",
|
||||
" if len(parts) != 2:\n",
|
||||
" raise ValueError(f\"Invalid S3 URI: {s3_uri}\")\n",
|
||||
" bucket, key = parts\n",
|
||||
" return bucket, key\n",
|
||||
"\n",
|
||||
"def construct_destination_key(bucket, key):\n",
|
||||
" \"\"\"\n",
|
||||
" Constructs the destination key by inserting 'ARCHIVE/' after the batch prefix.\n",
|
||||
"\n",
|
||||
" :param bucket: The S3 bucket name.\n",
|
||||
" :param key: The original S3 object key.\n",
|
||||
" :return: The destination S3 object key.\n",
|
||||
" \"\"\"\n",
|
||||
" # Split the key into parts\n",
|
||||
" parts = key.split('/', 1)\n",
|
||||
" if len(parts) == 1:\n",
|
||||
" # If there's no subfolder, place ARCHIVE directly\n",
|
||||
" archive_key = posixpath.join('ARCHIVE', parts[0])\n",
|
||||
" else:\n",
|
||||
" batch_prefix, sub_key = parts\n",
|
||||
" archive_key = posixpath.join(batch_prefix, 'ARCHIVE', sub_key)\n",
|
||||
" return archive_key\n",
|
||||
"\n",
|
||||
"def generate_mv_command(source_uri, destination_uri):\n",
|
||||
" \"\"\"\n",
|
||||
" Generates the aws s3 mv command.\n",
|
||||
"\n",
|
||||
" :param source_uri: Source S3 URI.\n",
|
||||
" :param destination_uri: Destination S3 URI.\n",
|
||||
" :return: The command as a list suitable for subprocess.\n",
|
||||
" \"\"\"\n",
|
||||
" command = [\n",
|
||||
" 'aws',\n",
|
||||
" 's3',\n",
|
||||
" 'mv',\n",
|
||||
" source_uri,\n",
|
||||
" destination_uri,\n",
|
||||
" '--profile',\n",
|
||||
" AWS_PROFILE,\n",
|
||||
" '--recursive'\n",
|
||||
" ]\n",
|
||||
" return command\n",
|
||||
"\n",
|
||||
"def execute_command(command):\n",
|
||||
" \"\"\"\n",
|
||||
" Executes a command using subprocess.\n",
|
||||
"\n",
|
||||
" :param command: The command as a list.\n",
|
||||
" :return: Tuple of (returncode, stdout, stderr)\n",
|
||||
" \"\"\"\n",
|
||||
" try:\n",
|
||||
" result = subprocess.run(\n",
|
||||
" command,\n",
|
||||
" check=False, # We'll handle errors manually\n",
|
||||
" stdout=subprocess.PIPE,\n",
|
||||
" stderr=subprocess.PIPE,\n",
|
||||
" text=True\n",
|
||||
" )\n",
|
||||
" return result.returncode, result.stdout, result.stderr\n",
|
||||
" except Exception as e:\n",
|
||||
" return -1, '', str(e)\n",
|
||||
"\n",
|
||||
"def main():\n",
|
||||
" \"\"\"\n",
|
||||
" Main function to process the list of S3 URIs and perform mv operations.\n",
|
||||
" \"\"\"\n",
|
||||
" log(f\"Starting S3 mv operations with DRY_RUN={DRY_RUN}\", 'info')\n",
|
||||
" \n",
|
||||
" for s3_uri in S3_URIS:\n",
|
||||
" try:\n",
|
||||
" bucket, key = parse_s3_uri(s3_uri)\n",
|
||||
" destination_key = construct_destination_key(bucket, key)\n",
|
||||
" destination_uri = f\"s3://{bucket}/{destination_key}\"\n",
|
||||
" \n",
|
||||
" log(f\"Source URI: {s3_uri}\", 'info')\n",
|
||||
" log(f\"Destination URI: {destination_uri}\", 'info')\n",
|
||||
" \n",
|
||||
" command = generate_mv_command(s3_uri, destination_uri)\n",
|
||||
" command_str = ' '.join(command)\n",
|
||||
" log(f\"Generated Command: {command_str}\", 'debug')\n",
|
||||
" \n",
|
||||
" if DRY_RUN:\n",
|
||||
" log(f\"[Dry Run] Command to execute: {command_str}\", 'info')\n",
|
||||
" continue # Skip execution in dry run mode\n",
|
||||
" \n",
|
||||
" # Execute the command\n",
|
||||
" returncode, stdout, stderr = execute_command(command)\n",
|
||||
" if returncode == 0:\n",
|
||||
" log(f\"Successfully moved: {s3_uri} to {destination_uri}\", 'info')\n",
|
||||
" else:\n",
|
||||
" log(f\"Failed to move: {s3_uri} to {destination_uri}\", 'error')\n",
|
||||
" log(f\"Error: {stderr.strip()}\", 'error')\n",
|
||||
" \n",
|
||||
" except Exception as e:\n",
|
||||
" log(f\"Error processing URI '{s3_uri}': {e}\", 'error')\n",
|
||||
" \n",
|
||||
" log(\"S3 mv operations completed.\", 'info')\n"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": 7,
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"Starting S3 mv operations with DRY_RUN=False\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/3b_missing_files_103024/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/3b_missing_files_103024/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/3b_missing_files_103024/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/3b_missing_files_103024/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/3b_missing_files_103024/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/3b_missing_files_103024/\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/dups_rerun_outputs/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/dups_rerun_outputs/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/dups_rerun_outputs/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/dups_rerun_outputs/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/dups_rerun_outputs/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/dups_rerun_outputs/\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/dups-for-textract/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/dups-for-textract/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/dups-for-textract/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/dups-for-textract/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/dups-for-textract/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/dups-for-textract/\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/for-textract/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/for-textract/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/for-textract/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/for-textract/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/for-textract/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/for-textract/\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/hotfix_rerun_112724/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/hotfix_rerun_112724/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/hotfix_rerun_112724/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/hotfix_rerun_112724/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/hotfix_rerun_112724/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/hotfix_rerun_112724/\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/redundant_dups/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/redundant_dups/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/redundant_dups/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/redundant_dups/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/redundant_dups/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/redundant_dups/\n",
|
||||
"Source URI: s3://centene-national-contracting-files/batch_3_priority_files/tables-for-textract/s3://centene-national-contracting-files/batch_3_priority_files/throttled_files/\n",
|
||||
"Destination URI: s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/tables-for-textract/s3://centene-national-contracting-files/batch_3_priority_files/throttled_files/\n",
|
||||
"DEBUG: Generated Command: aws s3 mv s3://centene-national-contracting-files/batch_3_priority_files/tables-for-textract/s3://centene-national-contracting-files/batch_3_priority_files/throttled_files/ s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/tables-for-textract/s3://centene-national-contracting-files/batch_3_priority_files/throttled_files/ --profile doczy_uat --recursive\n",
|
||||
"Successfully moved: s3://centene-national-contracting-files/batch_3_priority_files/tables-for-textract/s3://centene-national-contracting-files/batch_3_priority_files/throttled_files/ to s3://centene-national-contracting-files/batch_3_priority_files/ARCHIVE/tables-for-textract/s3://centene-national-contracting-files/batch_3_priority_files/throttled_files/\n",
|
||||
"S3 mv operations completed.\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
"source": [
|
||||
"main()"
|
||||
]
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"execution_count": null,
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"source": []
|
||||
}
|
||||
],
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"display_name": "Python 3",
|
||||
"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.10.4"
|
||||
}
|
||||
},
|
||||
"nbformat": 4,
|
||||
"nbformat_minor": 2
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
import boto3
|
||||
import pandas as pd
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
import os
|
||||
import math
|
||||
import csv
|
||||
|
||||
|
||||
"""
|
||||
This scirpt compares the file names in an Excel file (Batch excel files that MCS usually provides) with the file names in an S3 bucket.
|
||||
The Excel file must contain a column with the file names to search for.
|
||||
|
||||
This is used mainly to find files that are missing in the batch staging files and cannot be located in the s3 bucket.
|
||||
The output of this script can be then used with t_drive_search.py to find the missing files in the T drive.
|
||||
|
||||
Note: This script outputs both a CSV and an Excel file with the missing files and encoding issues.
|
||||
We use the output excel file with the tdrive search script to find the missing files in the T drive.
|
||||
|
||||
"""
|
||||
|
||||
# Function to list S3 objects and normalize the filenames
|
||||
def list_s3_files(bucket_name, prefix, s3_client):
|
||||
normalized_s3_files = []
|
||||
paginator = s3_client.get_paginator('list_objects_v2')
|
||||
for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
|
||||
if 'Contents' in page:
|
||||
for obj in page['Contents']:
|
||||
key = obj['Key']
|
||||
# Extract and normalize the filename (remove prefix and extension)
|
||||
# filename = key.split('/')[-1] # Take the file name from the key
|
||||
filename = os.path.basename(key) # Remove any directory path
|
||||
if len(filename) > 4:
|
||||
normalized_filename = filename[:-4] # Remove the last 4 characters (file extension)
|
||||
normalized_s3_files.append(normalized_filename)
|
||||
return set(normalized_s3_files)
|
||||
|
||||
# Main function for processing
|
||||
def find_missing_files(input_excel, bucket_name, prefix, output_csv, max_workers=10):
|
||||
# Initialize boto3 session and S3 client
|
||||
session = boto3.Session(profile_name='doczy_uat')
|
||||
s3_client = session.client('s3')
|
||||
|
||||
# Read the Excel file with header=1
|
||||
excel_df = pd.read_excel(input_excel, header=1, sheet_name='Sheet1')
|
||||
excel_filenames = excel_df['File Name'] # Remove extensions if present
|
||||
excel_filenames_set = set(excel_filenames)
|
||||
|
||||
# Get the normalized S3 filenames
|
||||
normalized_s3_files = list_s3_files(bucket_name, prefix, s3_client)
|
||||
|
||||
# Find files that are in Excel but not in S3
|
||||
missing_files = excel_filenames_set - normalized_s3_files
|
||||
|
||||
print(f"Type of missing_files: {type(missing_files)}")
|
||||
print(f"Type of normalized_files: {type(normalized_s3_files)}")
|
||||
print(f"Type of excel_filenames_set: {type(excel_filenames_set)}")
|
||||
|
||||
# Identify files that may have encoding differences, ensuring only strings are processed
|
||||
missing_files_list = [file for file in missing_files if isinstance(file, str) and not (isinstance(file, float) and math.isnan(file))]
|
||||
non_encoded_issues = [file for file in missing_files_list if all(ord(char) < 128 for char in file)]
|
||||
|
||||
output_df = pd.DataFrame({
|
||||
'Missing File Name': missing_files_list,
|
||||
'Encoding Check': ['No Encoding Issue' if file in non_encoded_issues else 'Encoding Issue' for file in missing_files_list]
|
||||
})
|
||||
|
||||
output_df.to_csv(output_csv, index=False, encoding='utf-8', quoting=csv.QUOTE_ALL)
|
||||
# Saving output as excel file
|
||||
output_df.to_excel(output_csv[:-4] + '.xlsx', index=False)
|
||||
|
||||
# Write the missing files and non-encoded issue files to the output CSV
|
||||
# with open(output_csv, mode='w', newline='', encoding='utf-8') as csvfile:
|
||||
# csvfile.write('Missing File Name,Encoding Check\n')
|
||||
# for file in missing_files_list:
|
||||
# encoding_status = 'No Encoding Issue' if file in non_encoded_issues else 'Encoding Issue'
|
||||
# csvfile.write(f"{file},{encoding_status}\n")
|
||||
|
||||
if __name__ == "__main__":
|
||||
input_excel = 'C:\\Doczy\\National contracting\\Allbatches\\batch_files\\updated_batches_102424\\first_batch\\Batch 14_10222024.xlsx' # Path to your input Excel file
|
||||
bucket_name = 'centene-national-contracting-files' # Your S3 bucket name
|
||||
prefix = 'batch_14_priority_files/TXT_FILES/' # Your S3 prefix
|
||||
output_csv = 'C:\\Doczy\\National contracting\\Allbatches\\diff_batch14_2.csv' # This will also save as xls
|
||||
find_missing_files(input_excel, bucket_name, prefix, output_csv)
|
||||
@@ -0,0 +1,187 @@
|
||||
import os
|
||||
import re
|
||||
import pandas as pd
|
||||
from pathlib import Path
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from functools import partial
|
||||
from tqdm import tqdm
|
||||
import logging
|
||||
|
||||
|
||||
"""
|
||||
This script searches for files in a directory based on a list of file names in an Excel file.
|
||||
The Excel file must contain a column with the file names to search for.
|
||||
|
||||
This is used mainly to find files that are missing in the batch staging files and cannot be located in the s3 bucket.
|
||||
|
||||
Usually the flow of staging and prepping batches involve:
|
||||
1. Staging the files in the s3 bucket
|
||||
2. Running the excel_s3_diff.py script to compare the files in the s3 bucket with the files in the staging folder
|
||||
3. The output of that script is an excel file that contains the missing files in the s3 bucket
|
||||
4. This script is then used to search for the missing files in the staging folder
|
||||
5. The output of this script is an excel file that contains the missing files in the staging folder
|
||||
6. The output from this diff will have files names with encoding issue / no enocoding issue.
|
||||
For files without any encoding issue usually are found in the t drive. We use the path to upoad it to a prefix in the s3 bucket for the client/batch_#
|
||||
7. Then the staged pdfs are then processed with the scheduler to run the pdfs through the textract pipeline to extract the text from the pdfs.
|
||||
|
||||
|
||||
"""
|
||||
|
||||
|
||||
|
||||
# --------------------------- Configuration ---------------------------
|
||||
|
||||
# Path to the Excel file
|
||||
# EXCEL_FILE_PATH = "C:\\Doczy\\National contracting\\Allbatches\\diff_batch13.xlsx" # <-- Update this path
|
||||
EXCEL_FILE_PATH = 'C:\\Doczy\\National contracting\\Allbatches\\new_diff_batch14_3_modified.xlsx'
|
||||
|
||||
# Directory to search within
|
||||
SEARCH_DIRECTORY = 'T:\\AArete Client Work\\Doczy-Production\\Restricted\\2024-06-28-pdf' # <-- Update this path
|
||||
|
||||
# Output Excel file path
|
||||
OUTPUT_FILE_PATH = 'C:\\Doczy\\National contracting\\Allbatches\\missing files analysis\\batch_14_search_output_3.xlsx' # <-- Update this path
|
||||
|
||||
# Maximum search depth
|
||||
MAX_DEPTH = 5
|
||||
|
||||
# Number of threads for multi-threading
|
||||
NUM_THREADS = 100 # Start with 4 and adjust as needed
|
||||
|
||||
# Log file path
|
||||
LOG_FILE_PATH = 'C:\\Doczy\\National contracting\\Allbatches\\missing files analysis\\batch_14_search_3.log'
|
||||
|
||||
# --------------------------- Logging Setup ---------------------------
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO, # Change to DEBUG for more detailed logs
|
||||
format='%(asctime)s [%(levelname)s] %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler(LOG_FILE_PATH),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
|
||||
# --------------------------- Helper Functions ---------------------------
|
||||
|
||||
def remove_suffix(file_name):
|
||||
"""
|
||||
Removes suffixes like (1), (2), etc., from the file name.
|
||||
Example: "XYZ_DOCUMENT(1).pdf" -> "XYZ_DOCUMENT.pdf"
|
||||
"""
|
||||
# Removing last 3 characters from the file name
|
||||
cleaned_name = file_name[:-3].strip() if file_name.endswith(('(1)', '(2)', '(3)', '(4)')) else file_name
|
||||
|
||||
logging.debug(f"Removed suffix: '{file_name}' -> '{cleaned_name}'")
|
||||
return cleaned_name
|
||||
|
||||
def find_files(search_dir, target_name, max_depth, file_types):
|
||||
"""
|
||||
Searches for files matching the target_name within search_dir up to max_depth.
|
||||
Returns a list of full file paths.
|
||||
"""
|
||||
logging.info(f"Searching for '{target_name}' in '{search_dir}' with max depth {max_depth}")
|
||||
results = []
|
||||
target_name_lower = target_name.lower()
|
||||
search_dir = Path(search_dir).resolve()
|
||||
base_depth = len(search_dir.parts)
|
||||
|
||||
for root, dirs, files in os.walk(search_dir, topdown=True):
|
||||
current_depth = len(Path(root).parts) - base_depth + 1
|
||||
if current_depth > max_depth:
|
||||
dirs[:] = [] # Prevent descending further
|
||||
logging.debug(f"Reached max depth at: {root}")
|
||||
continue
|
||||
|
||||
for file in files:
|
||||
if Path(file).suffix.lower() in file_types:
|
||||
if file.lower() == target_name_lower:
|
||||
file_path = Path(root) / file
|
||||
logging.info(f"Found file: {file_path}")
|
||||
results.append(file_path)
|
||||
|
||||
logging.info(f"Search complete for '{target_name}'. Found {len(results)} file(s).")
|
||||
return results
|
||||
|
||||
def search_file(row, search_dir, max_depth, file_types):
|
||||
"""
|
||||
Processes a single row to find the appropriate file path based on the Reason.
|
||||
"""
|
||||
# file_name = row['File Name']
|
||||
# reason = row['Reason']
|
||||
file_name = row['Missing File Name']
|
||||
reason = row['Encoding Check']
|
||||
found_path = None
|
||||
|
||||
# Remove suffix like (1), (2), etc.
|
||||
cleaned_file_name = remove_suffix(file_name)
|
||||
|
||||
# Ensure the file has a .pdf extension
|
||||
|
||||
cleaned_file_name += '.pdf'
|
||||
print(f"Cleaned file name: {cleaned_file_name}")
|
||||
|
||||
# Perform a case-insensitive search for the cleaned file name
|
||||
found_files = find_files(search_dir, cleaned_file_name, max_depth, file_types)
|
||||
|
||||
if found_files:
|
||||
# Select the file with the largest size if multiple are found
|
||||
largest_file = max(found_files, key=lambda f: f.stat().st_size)
|
||||
found_path = str(largest_file.resolve())
|
||||
logging.debug(f"Selected largest file: {found_path}")
|
||||
else:
|
||||
logging.warning(f"File '{cleaned_file_name}' not found for Reason: '{reason}'")
|
||||
|
||||
return found_path
|
||||
|
||||
# --------------------------- Main Processing ---------------------------
|
||||
|
||||
def main():
|
||||
logging.info("Script started.")
|
||||
|
||||
# Read the Excel file
|
||||
try:
|
||||
df = pd.read_excel(EXCEL_FILE_PATH, sheet_name='Sheet1', engine='openpyxl')
|
||||
logging.info(f"Excel file '{EXCEL_FILE_PATH}' read successfully.")
|
||||
except Exception as e:
|
||||
logging.error(f"Error reading Excel file: {e}")
|
||||
return
|
||||
|
||||
# Ensure required columns exist
|
||||
# required_columns = {'File Name', 'Reason'}
|
||||
required_columns = {'Missing File Name', 'Encoding Check'}
|
||||
if not required_columns.issubset(df.columns):
|
||||
logging.error(f"Error: The Excel sheet must contain the following columns: {required_columns}")
|
||||
return
|
||||
|
||||
# Initialize a new column for found paths
|
||||
df['Found Path'] = None
|
||||
|
||||
# Prepare for multi-threaded processing
|
||||
file_types = {'.pdf'}
|
||||
search_partial = partial(search_file, search_dir=SEARCH_DIRECTORY, max_depth=MAX_DEPTH, file_types=file_types)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
|
||||
# Submit all tasks
|
||||
futures = {executor.submit(search_partial, row): idx for idx, row in df.iterrows()}
|
||||
# Iterate through completed futures with a progress bar
|
||||
for future in tqdm(as_completed(futures), total=len(futures), desc="Processing"):
|
||||
idx = futures[future]
|
||||
try:
|
||||
found_path = future.result()
|
||||
df.at[idx, 'Found Path'] = found_path
|
||||
logging.debug(f"Row {idx} processed. Found Path: {found_path}")
|
||||
except Exception as e:
|
||||
df.at[idx, 'Found Path'] = f"Error: {e}"
|
||||
logging.error(f"Error processing row {idx}: {e}")
|
||||
|
||||
# Save the results to a new Excel file
|
||||
try:
|
||||
df.to_excel(OUTPUT_FILE_PATH, index=False)
|
||||
logging.info(f"Processing complete. Results saved to '{OUTPUT_FILE_PATH}'")
|
||||
except Exception as e:
|
||||
logging.error(f"Error saving output Excel file: {e}")
|
||||
|
||||
logging.info("Script finished.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user