import functools
import json
import re
import urllib.parse
from typing import Tuple

import pandas as pd
import requests
from sqlalchemy import create_engine, text
from trino.sqlalchemy import URL

from classes.data_operator import ExtendedDataFrame, ExtendedDataFrameList
from classes.secret import Secret
from scheduler.block import BaseBlock


@functools.lru_cache
def fetch_github_issue(access_token: str, owner: str, repo: str, issue: str) -> dict:
    url = f"https://api.github.com/repos/{owner}/{repo}/issues/{issue}"
    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(url, headers=headers)
    return response.json()


@functools.lru_cache
def search_google(cx: str, key: str, query: str) -> dict:
    query = urllib.parse.quote(query)
    url = f"https://www.googleapis.com/customsearch/v1?cx={cx}&key={key}&q={query}"
    return requests.get(url).json()


class GithubIssues(BaseBlock):
    def __init__(self, access_token: str, output_key: str = "body"):
        self.access_token = access_token
        self.output_key = output_key

    def __call__(self, links: list) -> list:
        contents = []
        for link in links:
            contents.append(self.get_content(link))
        return contents

    def get_content(self, link: str) -> str:
        item = self.parse_link(link)
        if item is None:
            return ""

        owner, repo, issue_number = item
        issue_data = fetch_github_issue(self.access_token, owner, repo, issue_number)
        for key in self.output_key.split("."):
            issue_data = issue_data.get(key, "")
            if not issue_data:
                return ""
        return issue_data

    def parse_link(self, link: str) -> Tuple[str, str, str]:
        pattern = r"https://github\.com/([^/]+)/([^/]+)/issues/(\d+)"
        match = re.match(pattern, link)
        if match:
            return match.group(1), match.group(2), match.group(3)


class GoogleSearch(BaseBlock):
    def __init__(
        self, cx: str, key: str, template: str, delimiter: str = "\n", top_k=5
    ):
        self.cx = cx
        self.key = key
        self.template = template
        self.delimiter = delimiter
        self.top_k = 5

    def __call__(self, texts: list) -> list:
        contents = []

        for text in texts:
            r = search_google(self.cx, self.key, text)
            contents.append(
                self.delimiter.join(
                    self.template.format(**item) for item in r["items"][: self.top_k]
                )
            )

        return contents


class ExtractRegex(BaseBlock):
    def __init__(self, pattern: str, flags: int = 0):
        self.pattern = pattern
        self.flags = flags

    def __call__(self, text: str) -> list:
        return re.findall(self.pattern, text, flags=self.flags)


class ReplaceRegex(BaseBlock):
    def __init__(self, pattern: str, replacement: str, count: int = 0, flags: int = 0):
        self.pattern = pattern
        self.replacement = replacement
        self.count = count
        self.flags = flags

    def __call__(self, text: str) -> str:
        return re.sub(
            self.pattern, self.replacement, text, count=self.count, flags=self.flags
        )


class DBQuery(BaseBlock):
    def __init__(
        self, host: str, user: str, password: Secret, schema: str, port: int = 4000
    ):
        db_url = f"""mysql+pymysql://{user}:{password}@{host}:{port}/{schema}?ssl=true&ssl_key=/etc/ssl/certs/ca-certificates.crt"""
        self.engine = create_engine(db_url)

    def __call__(self, query: str) -> ExtendedDataFrame:
        try:
            conn = self.engine.connect()
            df = pd.read_sql_query(text(query), conn)
            return ExtendedDataFrame(df)
        finally:
            conn.close()


class SQLExecutor(BaseBlock):
    def __init__(
        self, host: str, user: str, password: Secret, schema: str, port: int = 4000
    ):
        db_url = f"""mysql+pymysql://{user}:{password}@{host}:{port}/{schema}?ssl=true&ssl_key=/etc/ssl/certs/ca-certificates.crt"""
        self.engine = create_engine(db_url)

    def __call__(self, query: str) -> str:
        try:
            conn = self.engine.connect()
            conn.execute(text(query))
            return "Successful written"
        finally:
            conn.close()


class TrinoQuery(BaseBlock):
    def __init__(self, host: str, port: int = 8080):
        self.engine = create_engine(URL(host=host, port=port, user="admin"))

    def __call__(self, query: str) -> ExtendedDataFrame:
        try:
            conn = self.engine.connect()
            df = pd.read_sql_query(text(query), conn)
            return ExtendedDataFrame(df)
        finally:
            conn.close()


class BatchTrinoQuery(BaseBlock):
    def __init__(self, host: str, port: int = 8080):
        self.trino = TrinoQuery(host, port)

    def __call__(self, queries: list) -> ExtendedDataFrameList:
        return ExtendedDataFrameList([self.trino(query) for query in queries])