codedog_demo/callbacks.py

import time
import traceback
from functools import lru_cache
from os import environ as env
from os import listdir
from typing import List

import requests

from codedog_demo.github_utils import parse_github_pr_url

codedog_api = env.get("CODEDOG_API_URL", "")
github_token = env.get("GITHUB_TOKEN", "")

sample_names = []
sample_contents = []
for file in listdir("samples"):
    sample_names.append(file.replace("@", "/"))
    with open("samples/" + file, "r") as f:
        sample_contents.append(f.read())


def request_pr_review(url: str):
    try:
        repo, pr_number = parse_github_pr_url(url)
        if not repo or not pr_number:
            return "Invalid URL. Accept format is: https://www.github.com/{owner}/{repository}/pull/{pr_number}", ""

        result = _request_pr_review(repo, pr_number, ttl_hash=get_ttl_hash())

    except Exception:
        traceback.print_exc()
        return "Something went wrong. Please try again later.", ""
    return result, result


@lru_cache(maxsize=100)
def _request_pr_review(repo: str, pr_number: int, ttl_hash=None):
    response = requests.post(
        codedog_api, json={"repository": repo, "pull_request_number": pr_number, "token": github_token}
    )
    result = response.text
    if len(result) < 100:
        if result == "stream timeout":
            raise ValueError("Timeout")
        print(f"Error result: {result}")
        raise ValueError()
    return result


def get_ttl_hash(seconds=120):
    """Return the same value withing `seconds` time period"""
    return round(time.time() / seconds)


def get_sample_choices() -> List[str]:
    return sample_names


def show_sample(idx: int) -> str:
    return sample_contents[idx]