from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
from PyPDF2 import PdfReader

# Function to read data from a PDF file
def read_pdf():
    # Open the PDF file with your pdf file path
    with open('example.pdf', 'rb') as file: 
        # Create a PDF reader object
        pdf_reader = PdfReader(file)
        # Initialize variable to store data
        extracted_text = ""
        # Loop through each page and extract text
        for page in pdf_reader.pages:
            extracted_text += page.extract_text()
    
    return extracted_text

# Function to split text into two parts
def split_text(text=""):
    text_length = len(text)
    mid_index = text_length // 2
    return text[:mid_index], text[mid_index:]
# Function to write text to a file
def write_text_to_file(text="", file_name=""):
    with open(file_name, 'w') as file:
        file.write(text)
# Define default arguments for the DAG
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 2, 1),
    'schedule_interval': '@daily',
}
# Define the DAG with its name and default arguments
with DAG('pdf_data_transfer_workflow', default_args=default_args, catchup=False) as dag:
    # Task 1: Read PDF
    task_read_pdf = PythonOperator(
        task_id='read_pdf',
        python_callable=read_pdf
    )

    # Task 2: Split Text
    task_split_text = PythonOperator(
        task_id='split_text',
        python_callable=split_text,
        provide_context=True
    )

    # Task 3: Write Part 1 to File 
     task_write_part1 = PythonOperator(
        task_id='write_part1_to_file',
        python_callable=write_text_to_file,
        op_kwargs={'file_name': "text_file_part1.txt"},
        provide_context=True
    )
    # Task 4: Write Part 2 to File 

    task_write_part2 = PythonOperator(
        task_id='write_part2_to_file',
        python_callable=write_text_to_file,
        op_kwargs={'file_name': "text_file_part2.txt"},
        provide_context=True
    )
    # Define task dependencies
    task_read_pdf >> task_split_text >> [task_write_part1,task_write_part2]