# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

---
resources:

  - type: bigquery_table
    table_id: "drug_enforcement"
    description: "fda_drugspc"

dag:
  airflow_version: 2
  initialize:
    dag_id: drug_enforcement
    default_args:
      owner: "Google"
      depends_on_past: False
      start_date: '2021-03-01'
    max_active_runs: 1
    schedule_interval: "@daily"
    catchup: False
    default_view: graph

  tasks:

    - operator: "KubernetesPodOperator"
      description: "Run CSV transform within kubernetes pod"

      args:

        task_id: "transform_csv"
        name: "drug_enforcement"
        namespace: "composer-user-workloads"
        service_account_name: "default"
        config_file: "/home/airflow/composer_kube_config"
        image_pull_policy: "Always"
        image: "{{ var.json.fda_drug.container_registry.run_csv_transform_kub }}"
        env_vars:
          SOURCE_URL: "https://download.open.fda.gov/drug/enforcement/drug-enforcement-0001-of-0001.json.zip"
          SOURCE_FILE: "files/data.csv"
          TARGET_FILE: "files/data_output.csv"
          CHUNKSIZE: "50000"
          TARGET_GCS_BUCKET: "{{ var.value.composer_bucket }}"
          TARGET_GCS_PATH: "data/fda_drug/drug_enforcement/files/data_output.csv"
          TRANSFORM_LIST: >-
            [ "add_column", "rename_headers", "convert_date_format", "trim_whitespace", "reorder_headers", "replace_regex" ]
          REGEX_LIST: >-
            [
              [ "more_code_info", "null", "" ],
              [ "openfda_rxcui", "^(\\d+)$", ":$1" ]
            ]
          LOGGING_ENGLISH_NAME: "Food and Drug Administration (FDA) - Drug Enforcement"
          REORDER_HEADERS_LIST: >-
            [ "classification", "center_classification_date", "report_date", "postal_code", "termination_date",
              "recall_initiation_date", "recall_number", "city", "more_code_info", "event_id",
              "distribution_pattern", "openfda_application_number", "openfda_brand_name", "openfda_dosage_form", "openfda_generic_name",
              "openfda_manufacturer_name", "openfda_product_ndc", "openfda_product_type", "openfda_route", "openfda_substance_name",
              "openfda_spl_id", "openfda_spl_set_id", "openfda_pharm_class_moa", "openfda_pharm_class_cs", "openfda_pharm_class_pe",
              "openfda_pharm_class_epc", "openfda_upc", "openfda_unii", "openfda_rxcui", "recalling_firm",
              "voluntary_mandated", "state", "reason_for_recall", "initial_firm_notification", "status",
              "product_type", "country", "product_description", "code_info", "address_1",
              "address_2", "product_quantity" ]
          NEW_COLUMN_LIST: >-
            [
              ["openfda_dosage_form"]
            ]
          RENAME_HEADERS_LIST: >-
            {
              "classification": "classification",
              "center_classification_date": "center_classification_date",
              "report_date": "report_date",
              "postal_code": "postal_code",
              "termination_date": "termination_date",
              "recall_initiation_date": "recall_initiation_date",
              "recall_number": "recall_number",
              "city": "city",
              "more_code_info": "more_code_info",
              "event_id": "event_id",
              "distribution_pattern": "distribution_pattern",
              "openfda_application_number": "openfda_application_number",
              "openfda_brand_name": "openfda_brand_name",
              "openfda_generic_name": "openfda_generic_name",
              "openfda_manufacturer_name": "openfda_manufacturer_name",
              "openfda_product_ndc": "openfda_product_ndc",
              "openfda_product_type": "openfda_product_type",
              "openfda_route": "openfda_route",
              "openfda_substance_name": "openfda_substance_name",
              "openfda_spl_id": "openfda_spl_id",
              "openfda_spl_set_id": "openfda_spl_set_id",
              "openfda_pharm_class_moa": "openfda_pharm_class_moa",
              "openfda_pharm_class_cs": "openfda_pharm_class_cs",
              "openfda_pharm_class_pe": "openfda_pharm_class_pe",
              "openfda_pharm_class_epc": "openfda_pharm_class_epc",
              "openfda_upc": "openfda_upc",
              "openfda_unii": "openfda_unii",
              "openfda_rxcui": "openfda_rxcui",
              "recalling_firm": "recalling_firm",
              "voluntary_mandated": "voluntary_mandated",
              "state": "state",
              "reason_for_recall": "reason_for_recall",
              "initial_firm_notification": "initial_firm_notification",
              "status": "status",
              "product_type": "product_type",
              "country": "country",
              "product_description": "product_description",
              "code_info": "code_info",
              "address_1": "address_1",
              "address_2": "address_2",
              "product_quantity": "product_quantity",
              "openfda_dosage_form": "openfda_dosage_form"
            }
          DATE_FORMAT_LIST: >-
            [
              ["center_classification_date", "%Y%m%d", "%Y-%m-%d"],
              ["report_date", "%Y%m%d", "%Y-%m-%d"],
              ["termination_date", "%Y%m%d", "%Y-%m-%d"],
              ["recall_initiation_date", "%Y%m%d", "%Y-%m-%d"]
            ]
        container_resources:
          memory:
            request: "32Gi"
          cpu:
            request: "2"
          ephemeral-storage:
            request: "10Gi"

    - operator: "GoogleCloudStorageToBigQueryOperator"
      description: "Task to load CSV data to a BigQuery table"

      args:
        task_id: "load_to_bq"
        bucket: "{{ var.value.composer_bucket }}"
        source_objects: ["data/fda_drug/drug_enforcement/files/data_output.csv"]
        source_format: "CSV"
        destination_project_dataset_table: "{{ var.json.fda_drug.drug_enforcement_destination_table }}"
        skip_leading_rows: 1
        allow_quoted_newlines: True
        write_disposition: "WRITE_TRUNCATE"
        schema_fields:
          - "name": "classification"
            "type": "STRING"
            "description": "Numerical designation (I, II, or III) that is assigned by FDA to a particular product recall that indicates the relative degree of health hazard. Class I = Dangerous or defective products that predictably could cause serious health problems or death. Examples include: food found to contain botulinum toxin, food with undeclared allergens, a label mix-up on a lifesaving drug, or a defective artificial heart valve. Class II = Products that might cause a temporary health problem, or pose only a slight threat of a serious nature. Example: a drug that is under-strength but that is not used to treat life-threatening situations. Class III = Products that are unlikely to cause any adverse health reaction, but that violate FDA labeling or manufacturing laws. Examples include: a minor container defect and lack of English labeling in a retail food."
            "mode": "NULLABLE"
          - "name": "center_classification_date"
            "type": "DATE"
            "description": ""
            "mode": "NULLABLE"
          - "name": "report_date"
            "type": "DATE"
            "description": "Date that the FDA issued the enforcement report for the product recall."
            "mode": "NULLABLE"
          - "name": "postal_code"
            "type": "STRING"
            "description": ""
            "mode": "NULLABLE"
          - "name": "termination_date"
            "type": "DATE"
            "description": ""
            "mode": "NULLABLE"
          - "name": "recall_initiation_date"
            "type": "DATE"
            "description": "Date that the firm first began notifying the public or their consignees of the recall."
            "mode": "NULLABLE"
          - "name": "recall_number"
            "type": "STRING"
            "description": "A numerical designation assigned by FDA to a specific recall event used for tracking purposes."
            "mode": "NULLABLE"
          - "name": "city"
            "type": "STRING"
            "description": "The city in which the recalling firm is located."
            "mode": "NULLABLE"
          - "name": "more_code_info"
            "type": "STRING"
            "description": ""
            "mode": "NULLABLE"
          - "name": "event_id"
            "type": "INTEGER"
            "description": "A numerical designation assigned by FDA to a specific recall event used for tracking purposes."
            "mode": "NULLABLE"
          - "name": "distribution_pattern"
            "type": "STRING"
            "description": "General area of initial distribution such as, “Distributors in 6 states: NY, VA, TX, GA, FL and MA; the Virgin Islands; Canada and Japan”. The term “nationwide” is defined to mean the fifty states or a significant portion. Note that subsequent distribution by the consignees to other parties may not be included."
            "mode": "NULLABLE"
          - "name": "openfda_application_number"
            "type": "STRING"
            "description": "This corresponds to the NDA, ANDA, or BLA number reported by the labeler for products which have the corresponding Marketing Category designated. If the designated Marketing Category is OTC Monograph Final or OTC Monograph Not Final, then the application number will be the CFR citation corresponding to the appropriate Monograph (e.g. “part 341”). For unapproved drugs, this field will be null."
            "mode": "NULLABLE"
          - "name": "openfda_brand_name"
            "type": "STRING"
            "description": "The brand or trade name of the product."
            "mode": "NULLABLE"
          - "name": "openfda_dosage_form"
            "type": "STRING"
            "description": "The dosage form of the drug product."
            "mode": "NULLABLE"
          - "name": "openfda_generic_name"
            "type": "STRING"
            "description": "The dosage form of the drug product."
            "mode": "NULLABLE"
          - "name": "openfda_manufacturer_name"
            "type": "STRING"
            "description": "Name of company corresponding to the labeler code segment of the NDC."
            "mode": "NULLABLE"
          - "name": "openfda_product_ndc"
            "type": "STRING"
            "description": "The labeler manufacturer code and product code segments of the NDC number, separated by a hyphen."
            "mode": "NULLABLE"
          - "name": "openfda_product_type"
            "type": "STRING"
            "description": "The route of administration of the drug product."
            "mode": "NULLABLE"
          - "name": "openfda_route"
            "type": "STRING"
            "description": "The type of drug product."
            "mode": "NULLABLE"
          - "name": "openfda_substance_name"
            "type": "STRING"
            "description": "The list of active ingredients of a drug product."
            "mode": "NULLABLE"
          - "name": "openfda_spl_id"
            "type": "STRING"
            "description": "A unique identifier for a particular version of a Structured Product Label for a product. Also referred to as the document ID."
            "mode": "NULLABLE"
          - "name": "openfda_spl_set_id"
            "type": "STRING"
            "description": "A unique identifier for the Structured Product Label for a product, which is stable across versions of the label."
            "mode": "NULLABLE"
          - "name": "openfda_pharm_class_moa"
            "type": "STRING"
            "description": "Mechanism of action. Molecular, subcellular, or cellular level functional activity of a drug product’s pharmacologic class."
            "mode": "NULLABLE"
          - "name": "openfda_pharm_class_cs"
            "type": "STRING"
            "description": "Chemical structure. Chemical structure classification of a pharmacologic class."
            "mode": "NULLABLE"
          - "name": "openfda_pharm_class_pe"
            "type": "STRING"
            "description": "Physiologic effect. Tissue, organ, or organ system level functional activity of a pharmacologic class."
            "mode": "NULLABLE"
          - "name": "openfda_pharm_class_epc"
            "type": "STRING"
            "description": "Established pharmacologic class. A pharmacologic class associated with an approved indication of an active moiety that the FDA has determined to be scientifically valid and clinically meaningful."
            "mode": "NULLABLE"
          - "name": "openfda_upc"
            "type": "STRING"
            "description": "Documentation forthcoming."
            "mode": "NULLABLE"
          - "name": "openfda_unii"
            "type": "STRING"
            "description": "The Unique Ingredient Identifier of the drug or substance."
            "mode": "NULLABLE"
          - "name": "openfda_rxcui"
            "type": "STRING"
            "description": "The RxNorm Concept Unique Identifier. RxCUI is a unique number that describes a semantic concept about the drug product, including its ingredients, strength, and dosage forms."
            "mode": "NULLABLE"
          - "name": "recalling_firm"
            "type": "STRING"
            "description": "The firm that initiates a recall or, in the case of an FDA requested recall or FDA mandated recall, the firm that has primary responsibility for the manufacture and (or) marketing of the product to be recalled."
            "mode": "NULLABLE"
          - "name": "voluntary_mandated"
            "type": "STRING"
            "description": "Describes who initiated the recall. Recalls are almost always voluntary, meaning initiated by a firm. A recall is deemed voluntary when the firm voluntarily removes or corrects marketed products or the FDA requests the marketed products be removed or corrected. A recall is mandated when the firm was ordered by the FDA to remove or correct the marketed products, under section 518(e) of the FD&C Act, National Childhood Vaccine Injury Act of 1986, 21 CFR 1271.440, Infant Formula Act of 1980 and its 1986 amendments, or the Food Safety Modernization Act (FSMA)."
            "mode": "NULLABLE"
          - "name": "state"
            "type": "STRING"
            "description": "The U.S. state in which the recalling firm is located."
            "mode": "NULLABLE"
          - "name": "reason_for_recall"
            "type": "STRING"
            "description": "Information describing how the product is defective and violates the FD&C Act or related statutes."
            "mode": "NULLABLE"
          - "name": "initial_firm_notification"
            "type": "STRING"
            "description": "The method(s) by which the firm initially notified the public or their consignees of a recall. A consignee is a person or firm named in a bill of lading to whom or to whose order the product has or will be delivered."
            "mode": "NULLABLE"
          - "name": "status"
            "type": "STRING"
            "description": "On-Going = A recall which is currently in progress.  Completed = The recall action reaches the point at which the firm has actually retrieved and impounded all outstanding product that could reasonably be expected to be recovered, or has completed all product corrections. Terminated = FDA has determined that all reasonable efforts have been made to remove or correct the violative product in accordance with the recall strategy, and proper disposition has been made according to the degree of hazard. Pending = Actions that have been determined to be recalls, but that remain in the process of being classified."
            "mode": "NULLABLE"
          - "name": "product_type"
            "type": "STRING"
            "description": ""
            "mode": "NULLABLE"
          - "name": "country"
            "type": "STRING"
            "description": "The country in which the recalling firm is located."
            "mode": "NULLABLE"
          - "name": "product_description"
            "type": "STRING"
            "description": "Brief description of the product being recalled."
            "mode": "NULLABLE"
          - "name": "code_info"
            "type": "STRING"
            "description": "A list of all lot and/or serial numbers, product numbers, packer or manufacturer numbers, sell or use by dates, etc., which appear on the product or its labeling."
            "mode": "NULLABLE"
          - "name": "address_1"
            "type": "STRING"
            "description": ""
            "mode": "NULLABLE"
          - "name": "address_2"
            "type": "STRING"
            "description": ""
            "mode": "NULLABLE"
          - "name": "product_quantity"
            "type": "STRING"
            "description": "The amount of defective product subject to recall."
            "mode": "NULLABLE"
  graph_paths:
    - "transform_csv >> load_to_bq"
