import datetime
import io
import json
import jsonlines
import logging
import os
import pytz
import shutil
import sys
import tempfile
from flask import Request
from google.cloud import exceptions
from google.cloud import storage
from google.cloud import pubsub_v1
from google.api_core.exceptions import AlreadyExists
from pathlib import Path
from slack_bolt import App
from slack_sdk import errors
from typing import List

logfilename = 'ingest_log_at_{}.log'.format(datetime.datetime.now().isoformat())
logging.basicConfig(
    filename=logfilename,
    format='%(asctime)s %(message)s', datefmt='%Y-%m-%d %I:%M:%S %p',
    encoding='utf-8', level=logging.INFO)
TOPIC_INGEST_SLACK_TO_LAKE = os.environ.get("TOPIC_NAME")
PROJECT_ID = os.environ.get("PROJECT_ID")


def download_conversations_list(client, page_limit: int) -> List[dict]:
    """download Slack Web API conversations.list response.
        Returns:
            [{"id":xx, "name":yy}, {}, ...]
    """
    channels = []
    next_obj_exists = True
    next_cursor = None
    while next_obj_exists is True:
        slack_response = client.conversations_list(
                            cursor = next_cursor,
                            limit = page_limit, 
                            types = 'public_channel,private_channel')
        
        channels.extend(slack_response.get('channels'))
        next_cursor = slack_response.get('response_metadata').get('next_cursor')
        if next_cursor == "":
            next_obj_exists = False
    
    return channels


def download_users_list(client, page_limit: int) -> List[dict]:
    """download Slack Web API users.list response.
        Returns:
            [{"id":xx, "name":yy}, {}, ...]
    """
    users = []
    next_obj_exists = True
    next_cursor = None
    while next_obj_exists is True:
        slack_response = client.users_list(
                            cursor = next_cursor,
                            limit = page_limit)
        
        users.extend(slack_response.get('members'))
        next_cursor = slack_response.get('response_metadata').get('next_cursor')
        if next_cursor == "":
            next_obj_exists = False
    
    return users


def download_conversations_history(
    client, channel: str, page_limit: int,
    latest_unix_time: float, oldest_unix_time: float) -> List[dict]:
    """download Slack Web API conversations.list response.
        Returns:
            List of dict{"channel":ccc, "message":{ ... }}
    """
    conversations_by_channel = []
    next_obj_exists = True
    next_cursor = None
    while next_obj_exists is True:
        try:
            slack_response = client.conversations_history(
                                channel = channel,
                                cursor = next_cursor,
                                limit = page_limit,
                                latest = latest_unix_time,
                                oldest = oldest_unix_time)
            cnv_by_ch = slack_response.get('messages')
            for item in cnv_by_ch:
                item.update( {"channel": channel})
            conversations_by_channel.extend(cnv_by_ch)
            if slack_response.get('has_more') is False:
                next_cursor = ""
            else:
                next_cursor = slack_response.get('response_metadata').get('next_cursor')
        except errors.SlackApiError as e:
            logging.info(e)
            break
        if next_cursor == "":
            next_obj_exists = False
    
    return conversations_by_channel


def target_channel_id_name_list(
    conversations_list: list=None, including_archived: bool=False):
    """extract targeted channels id list from conversations_list response.
        Returns:
            id_list, name_list
    """
    id_list = []
    name_list = []
    for ch in conversations_list:
        if including_archived is False:
            if ch['is_archived'] is True:
                continue
        id_list.append(ch['id'])
        name_list.append(ch['name'])
    return id_list, name_list


def exporting_dir(oldest_ut: float=None) -> str:
    oldest_dt = datetime.datetime.fromtimestamp(oldest_ut)
    oldest_dt_str = datetime.datetime.strftime(oldest_dt, format='%Y-%m-%d')
    dir_name = "slack_lake/daily-ingest_target-date_{}".format(oldest_dt_str)
    dir_path = '{}'.format(dir_name)
    #Path(dir_path).mkdir(parents=True, exist_ok=True)
    return dir_path


# ==  BEGIN - Main Cloud Function  ==
def ingest_slack_data(request, **kwargs):
    """ingest slack data + publish topic
        Arguments:
            - request (flask.Request): The request object.
            - **kwargs for locally test
            latest_ut: float=None, oldest_ut: float=None, bucket_name: str=None
            - **kwargs[latest_ut]: float
            - **kwargs[oldest_ut]: float
            - **kwargs[bucket_name]: str
        * Request Body
            latest_ut(float): データ取得対象の最新タイムスタンプ（UNIXタイム）
            oldest_ut: データ取得対象の最古タイムスタンプ（UNIXタイム）
            bucket_name: 保存先のバケット名
    """
    # ボットトークンと署名シークレットを使ってアプリを初期化します
    app = App(
        # process_before_response must be True when running on FaaS
        process_before_response=True,
        token=os.environ.get("SLACK_BOT_TOKEN"),
        signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
    )
    
    # Settings
    _conditions = {}
    if request is None: # locally test
        _conditions = kwargs
    else : # production environment
        req_data = request.get_json()
        _conditions = {} if req_data is None else req_data
    print('■ request json data\n', _conditions)
    latest_unix_time = _conditions['latest_ut'] if ('latest_ut' in _conditions.keys()) else None
    oldest_unix_time = _conditions['oldest_ut'] if ('oldest_ut' in _conditions.keys()) else None
    bucket_name = _conditions['bucket_name'] if ('bucket_name' in _conditions.keys()) else None
    
    # 時刻が明示されていない場合は、通常のデイリー実行を前提として
    # データ取得期間を定義する
    if latest_unix_time is None or oldest_unix_time is None:
        tz = pytz.timezone('Asia/Tokyo')
        start_of_today = datetime.datetime.now(tz).replace(hour=0,minute=0,second=0,microsecond=0)
        latest_unix_time = start_of_today.timestamp()
        start_of_yesterday = start_of_today - datetime.timedelta(days=1)
        oldest_unix_time = start_of_yesterday.timestamp()
    out_dir = exporting_dir(oldest_ut=oldest_unix_time)
    logging.info('out_dir : {}'.format(out_dir))
    logging.info('oldest_ut : {}'.format(oldest_unix_time) + ' | latest_ut : {}'.format(latest_unix_time))
    
    client = app.client

    storage_client = storage.Client()
    if bucket_name is None:
        bucket_name = "dl-guild-slack-data"
    bucket = None
    try:
        bucket = storage_client.bucket(bucket_name)
    except exceptions.GoogleCloudError as e:
        logging.warning('On create bucket object for {}'.format(bucket_name))    
        logging.warning(e)
    
    
    # ingest channles list
    channels = download_conversations_list(client=client, page_limit=100)
    save_into_bucket(channels, bucket, out_dir + '/' + 'conversations_list.json')
    
    # ingest users list
    users = download_users_list(client=client, page_limit=100)
    save_into_bucket(users, bucket, out_dir + '/' + 'users_list.json')
    
    # ingest conversations history
    channel_id_list, channel_name_list = target_channel_id_name_list(channels, including_archived=False)
    conversations = []
    for channel_id, channel_name in zip(channel_id_list, channel_name_list):
        logging.info('download conversations (ch_id: {0}, ch_name: {1})'.format(
            channel_id, channel_name))
        conversations_by_ch = download_conversations_history(
            client=client, channel=channel_id, page_limit=100, latest_unix_time=latest_unix_time, oldest_unix_time=oldest_unix_time
        )
        if len(conversations_by_ch) > 0:
            conversations.extend(conversations_by_ch)
    save_into_bucket(conversations, bucket, out_dir + '/' + 'conversations_history.json')
    
    # save completing log
    tz = pytz.timezone('Asia/Tokyo')
    now = datetime.datetime.now(tz)
    ingest_log = {'ingested_at_ts': now.timestamp(), 'ingested_at': now.strftime('%Y-%m-%d %H:%M:%S')}
    save_into_bucket(ingest_log, bucket, out_dir + '/' + 'ingest_log.json')
    
    # ----------------------------------------------------
    # publish topic to trigger Loading to BQ function
    publisher = pubsub_v1.PublisherClient()
    topic_name = TOPIC_INGEST_SLACK_TO_LAKE
    topic_path = publisher.topic_path(PROJECT_ID, topic_name)
    
    # create topic if not exists
    try:
        response = publisher.create_topic(request={"name": topic_path})
        logging.info(f"Created a topic\n{response}")
    except AlreadyExists:
        logging.info(f"{topic_name} already exists.")
    
    # publishes a message
    msg_bytes = json.dumps({
        'data': {
            'message': 'ingested slack data into cloud storage as jsonl files.',
            'blob-dir-path': out_dir
        }
    }).encode('utf-8')
    pub_result = ''
    try:
        publish_future = publisher.publish(topic_path, data=msg_bytes)
        publish_future.result()  # Verify the publish succeeded
        pub_result = f"Message published \ntopic : {TOPIC_INGEST_SLACK_TO_LAKE}\nmsg : key=blob-dir-path / val={out_dir}"
        logging.info(pub_result)
    except Exception as e:
        logging.error(e)

    # == only local env ==
    # # copy log to export dir
    # from_log_path = Path(logfilename)
    # to_log_path = Path(out_dir)
    # shutil.copy2(from_log_path, to_log_path)
    # == only local env ==
    
    return f"Successfully ingested slack data.\n{pub_result}"
# ==  END - Main Cloud Function  ==


# ==  BEGIN - Sub Cloud Function  ==
def save_into_bucket(data: List[dict], bucket: storage.bucket.Bucket, obj_name: str=None):
    """save response data as json into cloud storage bucket
    """
    blob = storage.blob.Blob(name=obj_name, bucket=bucket)
    f = io.BytesIO()
    with jsonlines.Writer(f) as writer:
        for d in data:
            writer.write(d)
    try:
        blob.upload_from_file(f, rewind=True)
    except exceptions.GoogleCloudError as e:
        logging.warning('Upload Error : {}'.format(obj_name))
        logging.warning(e)
    logging.info('save {}'.format(obj_name))
    f.close()
# ==  END - Sub Cloud Function  ==


# ==  BEGIN - Sub Cloud Function for Test  ==
def save_as_json(data: List[dict], fname: str=None):
    """save response data as json
    """
    with open(fname, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=4)
        logging.info('save {}'.format(fname))
# ==  END - Sub Cloud Function for Test  ==


# run app
if __name__ == "__main__":
    # parse args
    args = sys.argv
    latest_ut = 0
    oldest_ut = 0
    if len(args) > 2:
        latest_ut = float(args[1])
        oldest_ut = float(args[2])
    # main proc
    return_str = ingest_slack_data(None, latest_ut=latest_ut, oldest_ut=oldest_ut)
    logging.info(return_str)

import datetime
import io
import json
import jsonlines
import logging
import os
import pytz
import shutil
import sys
import tempfile
from flask import Request
from google.cloud import exceptions
from google.cloud import storage
from google.cloud import pubsub_v1
from google.api_core.exceptions import AlreadyExists
from pathlib import Path
from slack_bolt import App
from slack_sdk import errors
from typing import List

logfilename = 'ingest_log_at_{}.log'.format(datetime.datetime.now().isoformat())
logging.basicConfig(
    filename=logfilename,
    format='%(asctime)s %(message)s', datefmt='%Y-%m-%d %I:%M:%S %p',
    encoding='utf-8', level=logging.INFO)
TOPIC_INGEST_SLACK_TO_LAKE = os.environ.get("TOPIC_NAME")
PROJECT_ID = os.environ.get("PROJECT_ID")


def download_conversations_list(client, page_limit: int) -> List[dict]:
    """download Slack Web API conversations.list response.
        Returns:
            [{"id":xx, "name":yy}, {}, ...]
    """
    channels = []
    next_obj_exists = True
    next_cursor = None
    while next_obj_exists is True:
        slack_response = client.conversations_list(
                            cursor = next_cursor,
                            limit = page_limit, 
                            types = 'public_channel,private_channel')
        
        channels.extend(slack_response.get('channels'))
        next_cursor = slack_response.get('response_metadata').get('next_cursor')
        if next_cursor == "":
            next_obj_exists = False
    
    return channels


def download_users_list(client, page_limit: int) -> List[dict]:
    """download Slack Web API users.list response.
        Returns:
            [{"id":xx, "name":yy}, {}, ...]
    """
    users = []
    next_obj_exists = True
    next_cursor = None
    while next_obj_exists is True:
        slack_response = client.users_list(
                            cursor = next_cursor,
                            limit = page_limit)
        
        users.extend(slack_response.get('members'))
        next_cursor = slack_response.get('response_metadata').get('next_cursor')
        if next_cursor == "":
            next_obj_exists = False
    
    return users


def download_conversations_history(
    client, channel: str, page_limit: int,
    latest_unix_time: float, oldest_unix_time: float) -> List[dict]:
    """download Slack Web API conversations.list response.
        Returns:
            List of dict{"channel":ccc, "message":{ ... }}
    """
    conversations_by_channel = []
    next_obj_exists = True
    next_cursor = None
    while next_obj_exists is True:
        try:
            slack_response = client.conversations_history(
                                channel = channel,
                                cursor = next_cursor,
                                limit = page_limit,
                                latest = latest_unix_time,
                                oldest = oldest_unix_time)
            cnv_by_ch = slack_response.get('messages')
            for item in cnv_by_ch:
                item.update( {"channel": channel})
            conversations_by_channel.extend(cnv_by_ch)
            if slack_response.get('has_more') is False:
                next_cursor = ""
            else:
                next_cursor = slack_response.get('response_metadata').get('next_cursor')
        except errors.SlackApiError as e:
            logging.info(e)
            break
        if next_cursor == "":
            next_obj_exists = False
    
    return conversations_by_channel


def target_channel_id_name_list(
    conversations_list: list=None, including_archived: bool=False):
    """extract targeted channels id list from conversations_list response.
        Returns:
            id_list, name_list
    """
    id_list = []
    name_list = []
    for ch in conversations_list:
        if including_archived is False:
            if ch['is_archived'] is True:
                continue
        id_list.append(ch['id'])
        name_list.append(ch['name'])
    return id_list, name_list


def exporting_dir(oldest_ut: float=None) -> str:
    oldest_dt = datetime.datetime.fromtimestamp(oldest_ut)
    oldest_dt_str = datetime.datetime.strftime(oldest_dt, format='%Y-%m-%d')
    dir_name = "slack_lake/daily-ingest_target-date_{}".format(oldest_dt_str)
    dir_path = '{}'.format(dir_name)
    #Path(dir_path).mkdir(parents=True, exist_ok=True)
    return dir_path


# ==  BEGIN - Main Cloud Function  ==
def ingest_slack_data(request, **kwargs):
    """ingest slack data + publish topic
        Arguments:
            - request (flask.Request): The request object.
            - **kwargs for locally test
            latest_ut: float=None, oldest_ut: float=None, bucket_name: str=None
            - **kwargs[latest_ut]: float
            - **kwargs[oldest_ut]: float
            - **kwargs[bucket_name]: str
        * Request Body
            latest_ut(float): データ取得対象の最新タイムスタンプ（UNIXタイム）
            oldest_ut: データ取得対象の最古タイムスタンプ（UNIXタイム）
            bucket_name: 保存先のバケット名
    """
    # ボットトークンと署名シークレットを使ってアプリを初期化します
    app = App(
        # process_before_response must be True when running on FaaS
        process_before_response=True,
        token=os.environ.get("SLACK_BOT_TOKEN"),
        signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
    )
    
    # Settings
    _conditions = {}
    if request is None: # locally test
        _conditions = kwargs
    else : # production environment
        req_data = request.get_json()
        _conditions = {} if req_data is None else req_data
    print('■ request json data\n', _conditions)
    latest_unix_time = _conditions['latest_ut'] if ('latest_ut' in _conditions.keys()) else None
    oldest_unix_time = _conditions['oldest_ut'] if ('oldest_ut' in _conditions.keys()) else None
    bucket_name = _conditions['bucket_name'] if ('bucket_name' in _conditions.keys()) else None
    
    # 時刻が明示されていない場合は、通常のデイリー実行を前提として
    # データ取得期間を定義する
    if latest_unix_time is None or oldest_unix_time is None:
        tz = pytz.timezone('Asia/Tokyo')
        start_of_today = datetime.datetime.now(tz).replace(hour=0,minute=0,second=0,microsecond=0)
        latest_unix_time = start_of_today.timestamp()
        start_of_yesterday = start_of_today - datetime.timedelta(days=1)
        oldest_unix_time = start_of_yesterday.timestamp()
    out_dir = exporting_dir(oldest_ut=oldest_unix_time)
    logging.info('out_dir : {}'.format(out_dir))
    logging.info('oldest_ut : {}'.format(oldest_unix_time) + ' | latest_ut : {}'.format(latest_unix_time))
    
    client = app.client

    storage_client = storage.Client()
    if bucket_name is None:
        bucket_name = "dl-guild-slack-data"
    bucket = None
    try:
        bucket = storage_client.bucket(bucket_name)
    except exceptions.GoogleCloudError as e:
        logging.warning('On create bucket object for {}'.format(bucket_name))    
        logging.warning(e)
    
    
    # ingest channles list
    channels = download_conversations_list(client=client, page_limit=100)
    save_into_bucket(channels, bucket, out_dir + '/' + 'conversations_list.json')
    
    # ingest users list
    users = download_users_list(client=client, page_limit=100)
    save_into_bucket(users, bucket, out_dir + '/' + 'users_list.json')
    
    # ingest conversations history
    channel_id_list, channel_name_list = target_channel_id_name_list(channels, including_archived=False)
    conversations = []
    for channel_id, channel_name in zip(channel_id_list, channel_name_list):
        logging.info('download conversations (ch_id: {0}, ch_name: {1})'.format(
            channel_id, channel_name))
        conversations_by_ch = download_conversations_history(
            client=client, channel=channel_id, page_limit=100, latest_unix_time=latest_unix_time, oldest_unix_time=oldest_unix_time
        )
        if len(conversations_by_ch) > 0:
            conversations.extend(conversations_by_ch)
    save_into_bucket(conversations, bucket, out_dir + '/' + 'conversations_history.json')
    
    # save completing log
    tz = pytz.timezone('Asia/Tokyo')
    now = datetime.datetime.now(tz)
    ingest_log = {'ingested_at_ts': now.timestamp(), 'ingested_at': now.strftime('%Y-%m-%d %H:%M:%S')}
    save_into_bucket(ingest_log, bucket, out_dir + '/' + 'ingest_log.json')
    
    # ----------------------------------------------------
    # publish topic to trigger Loading to BQ function
    publisher = pubsub_v1.PublisherClient()
    topic_name = TOPIC_INGEST_SLACK_TO_LAKE
    topic_path = publisher.topic_path(PROJECT_ID, topic_name)
    
    # create topic if not exists
    try:
        response = publisher.create_topic(request={"name": topic_path})
        logging.info(f"Created a topic\n{response}")
    except AlreadyExists:
        logging.info(f"{topic_name} already exists.")
    
    # publishes a message
    msg_bytes = json.dumps({
        'data': {
            'message': 'ingested slack data into cloud storage as jsonl files.',
            'blob-dir-path': out_dir
        }
    }).encode('utf-8')
    pub_result = ''
    try:
        publish_future = publisher.publish(topic_path, data=msg_bytes)
        publish_future.result()  # Verify the publish succeeded
        pub_result = f"Message published \ntopic : {TOPIC_INGEST_SLACK_TO_LAKE}\nmsg : key=blob-dir-path / val={out_dir}"
        logging.info(pub_result)
    except Exception as e:
        logging.error(e)

    # == only local env ==
    # # copy log to export dir
    # from_log_path = Path(logfilename)
    # to_log_path = Path(out_dir)
    # shutil.copy2(from_log_path, to_log_path)
    # == only local env ==
    
    return f"Successfully ingested slack data.\n{pub_result}"
# ==  END - Main Cloud Function  ==


# ==  BEGIN - Sub Cloud Function  ==
def save_into_bucket(data: List[dict], bucket: storage.bucket.Bucket, obj_name: str=None):
    """save response data as json into cloud storage bucket
    """
    blob = storage.blob.Blob(name=obj_name, bucket=bucket)
    f = io.BytesIO()
    with jsonlines.Writer(f) as writer:
        for d in data:
            writer.write(d)
    try:
        blob.upload_from_file(f, rewind=True)
    except exceptions.GoogleCloudError as e:
        logging.warning('Upload Error : {}'.format(obj_name))
        logging.warning(e)
    logging.info('save {}'.format(obj_name))
    f.close()
# ==  END - Sub Cloud Function  ==


# ==  BEGIN - Sub Cloud Function for Test  ==
def save_as_json(data: List[dict], fname: str=None):
    """save response data as json
    """
    with open(fname, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=4)
        logging.info('save {}'.format(fname))
# ==  END - Sub Cloud Function for Test  ==


# run app
if __name__ == "__main__":
    # parse args
    args = sys.argv
    latest_ut = 0
    oldest_ut = 0
    if len(args) > 2:
        latest_ut = float(args[1])
        oldest_ut = float(args[2])
    # main proc
    return_str = ingest_slack_data(None, latest_ut=latest_ut, oldest_ut=oldest_ut)
    logging.info(return_str)

download Slack Web API conversations.list response.         Returns:             [{"id":xx, "name":yy}, {}, ...] download Slack Web API users.list response.         Returns:             [{"id":xx, "name":yy}, {}, ...] download Slack Web API conversations.list response.         Returns:             List of dict{"channel":ccc, "message":{ ... }} extract targeted channels id list from conversations_list response.         Returns:             id_list, name_list #Path(dir_path).mkdir(parents=True, exist_ok=True) # ==  BEGIN - Main Cloud Function  == ingest slack data + publish topic         Arguments:             - request (flask.Request): The request object.             - **kwargs for locally test             latest_ut: float=None, oldest_ut: float=None, bucket_name: str=None             - **kwargs[latest_ut]: float             - **kwargs[oldest_ut]: float             - **kwargs[bucket_name]: str         * Request Body             latest_ut(float): データ取得対象の最新タイムスタンプ（UNIXタイム）             oldest_ut: データ取得対象の最古タイムスタンプ（UNIXタイム）             bucket_name: 保存先のバケット名 # ボットトークンと署名シークレットを使ってアプリを初期化します # process_before_response must be True when running on FaaS # Settings # locally test # production environment # 時刻が明示されていない場合は、通常のデイリー実行を前提として # データ取得期間を定義する # ingest channles list # ingest users list # ingest conversations history # save completing log # ---------------------------------------------------- # publish topic to trigger Loading to BQ function # create topic if not exists # publishes a message # Verify the publish succeeded # == only local env == # # copy log to export dir # from_log_path = Path(logfilename) # to_log_path = Path(out_dir) # shutil.copy2(from_log_path, to_log_path) # == only local env == # ==  END - Main Cloud Function  == # ==  BEGIN - Sub Cloud Function  == save response data as json into cloud storage bucket # ==  END - Sub Cloud Function  == # ==  BEGIN - Sub Cloud Function for Test  == save response data as json # ==  END - Sub Cloud Function for Test  == # run app # parse args # main proc