#!/usr/bin/env python
# Copyright 2020 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.

# A script to detect all pipelines declared in pipelines/ and cache/ and
# dump a summary of their configurations within a `tmp/` folder in the
# project root.
#
# Example usage: `python src/scripts/list_pipelines.py`

import sys
import os
import json

path = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.append(path)

from pandas import DataFrame
from lib.constants import SRC
from lib.pipeline import DataPipeline
from typing import List, Iterator, Dict


def get_pipeline_names() -> Iterator[str]:
    for item in (SRC / "pipelines").iterdir():
        if not item.name.startswith("_") and not item.is_file():
            yield item.name


def get_source_configs(pipeline_names: List[str]) -> Iterator[Dict]:
    """Map a list of pipeline names to their source configs."""

    for pipeline_name in pipeline_names:
        data_pipeline = DataPipeline.load(pipeline_name)

        for data_source in data_pipeline.data_sources:
            data_source_config = data_source.config

            data_source_name = data_source_config.get("name")
            data_source_fetch_params = data_source_config.get("fetch", [])

            for fetch_param in data_source_fetch_params:
                file_ext = fetch_param.get("opts", {}).get("ext", "")
                data_source_url = fetch_param.get("url")

                yield {
                    "pipeline_name": pipeline_name,
                    "source_name": data_source_name,
                    "url": data_source_url,
                    "ext": file_ext,
                }


def get_cache_configs() -> Iterator[Dict]:
    """Generate a list of configurations for cached data sources."""

    config_json = SRC / "cache" / "config.json"
    with open(config_json, "r") as fd:
        cache_pipelines_config = json.load(fd)
        for pipeline_config in cache_pipelines_config:
            yield {"url": pipeline_config.get("url"), "cmd": pipeline_config.get("cmd")}


if __name__ == "__main__":
    source_configs_df = DataFrame(get_source_configs(get_pipeline_names()))
    cache_configs_df = DataFrame(get_cache_configs())

    print(source_configs_df)
    print(cache_configs_df)

    source_configs_df.to_csv(SRC / ".." / "tmp" / "source_configs.csv")
    cache_configs_df.to_csv(SRC / ".." / "tmp" / "cache_configs.csv")

@@ -937,16 +937,28 @@
 (path)%0A%0A
+import yaml%0A
 from pan
@@ -1068,14 +1068,8 @@
 port
- List,
  Ite
@@ -1310,12 +1310,16 @@
 es: 
-List
+Iterator
 %5Bstr
@@ -1636,20 +1636,21 @@
 _source_
-name
+class
  = data_
@@ -1668,21 +1668,91 @@
 ig.get(%22
-name%22
+class%22)%0A            data_source_uuid = data_source.uuid(data_pipeline.table
 )%0A      
@@ -2055,37 +2055,34 @@
    %22
-pipeline_name%22: pipeline_name
+source%22: data_source_class
 ,%0A  
@@ -2100,27 +2100,19 @@
        %22
-source_name
+url
 %22: data_
@@ -2118,20 +2118,19 @@
 _source_
-name
+url
 ,%0A      
@@ -2137,34 +2137,35 @@
               %22u
-rl
+uid
 %22: data_source_u
@@ -2156,34 +2156,35 @@
 %22: data_source_u
-rl
+uid
 ,%0A              
@@ -2344,26 +2344,27 @@
 %22%0A%0A    c
+ache_c
 onfig
-_json
  = SRC /
@@ -2374,24 +2374,13 @@
 ache
-%22 / %22config.json
+.yaml
 %22%0A  
@@ -2395,19 +2395,20 @@
 pen(
+cache_
 config
-_json
 , %22r
@@ -2429,190 +2429,37 @@
     
-cache_pipelines_config = json.load(fd)%0A        for pipeline_config in cache_pipelines_config:%0A            yield %7B%22url%22: pipeline_config.get(%22url%22), %22cmd%22: pipeline_config.get(%22cmd%22)%7D
+yield from yaml.safe_load(fd)
 %0A%0A%0Ai
@@ -2646,161 +2646,42 @@
 s_df
-)%0A    print(cache_configs_df)%0A%0A    source_configs_df.to_csv(SRC / %22..%22 / %22tmp%22 / %22source_configs.csv%22)%0A    cache_configs_df.to_csv(SRC / %22..%22 / %22tmp%22 / %22
+.to_csv(index=False))%0A    # print(
 cach
@@ -2693,11 +2693,9 @@
 figs
-.csv%22
+_df
 )%0A

Fix list_pipelines script