Composer DAG for Data Ingestion Pipeline

I need a Composer DAG to ingest data from CSV, XLS, and TXT files, there are like 25 processes for which we receive data . There are some pre ingestion steps involved , like running python scripts, the dag should identify the which python script it should run for the given input file.if the file is in xls format then it should run a different python script which converts xls to csv.The ingested data should be stored in BQ.
The dag also should trigger dataflow job with input parameters.

Requirements:
- Create a Composer DAG for data ingestion
- Support input from CSV, XLS, TXT formats and run pre ingestion different python scripts available for each process
- event based trigger using pubsub
- Store ingested data in bigquery

Ideal Skills and Experience:
- Expertise in Google Cloud Composer
- Experience with data ingestion from various file formats
- Knowledge of cloud storage solutions