<commit_before>defmodule Krihelinator.Periodic do
  require Logger
  import Ecto.Query, only: [from: 2]
  alias Krihelinator.Repo
  alias Krihelinator.Github, as: GH

  @moduledoc """
  A background task that needs to run periodically on the Krihelinator server.

  - Mark all github repos as "dirty".
  - Scrape the github trending page for interesting, active, projects.
  - Using BigQuery, get and scrape all repos that at least 2 users pushed
    commits to.
  - Get the remaining "dirty" repos and pass through the scraper again, to
    update stats.
  - Clean the remaining dirty repos. These repos failed to update or fell bellow
    activity threshold.
  - Update the total krihelimeter for all languages.
  - Scrape showcases from github and update repos that belongs to showcases.
  """

  defp periodically_gc(pid) do
    # This is an ugly hack to force GC every now and then on the periodic
    # process. TODO search for a better solution!
    Process.sleep(30_000)
    if Process.alive?(pid) do
      :erlang.garbage_collect(pid)
      periodically_gc(pid)
    end
  end

  @doc """
  Main entry point.
  """
  def run() do
    my_pid = self()
    Task.async(fn -> periodically_gc(my_pid) end)
    Logger.info "Periodic process kicked in!"
    set_dirty_bit()
    scrape_trending()
    scrape_from_bigquery()
    rescrape_still_dirty()
    clean_dirty()
    update_languages_stats()
    scrape_showcases()
    Logger.info "Periodic process finished successfully!"
  end

  def set_dirty_bit() do
    Logger.info "Setting dirty bit for all"
    Repo.update_all(GH.Repo, set: [dirty: true])
  end

  @doc """
  Scrape the github trending page and update repos.
  """
  def scrape_trending() do
    Logger.info "Resetting trendiness for all"
    Repo.update_all(GH.Repo, set: [trending: false])
    Logger.info "Scraping trending"
    Krihelinator.Periodic.GithubTrending.scrape()
    |> async_scrape()
    |> Stream.map(fn
      {:ok, data} -> {:ok, Map.put(data, :trending, true)}
      otherwise -> otherwise
    end)
    |> handle_scraped_data()
  end

  @doc """
  Request repos from google BigQuery, scrape, and persist.
  """
  def scrape_from_bigquery() do
    Logger.info "Getting repositories from BigQuery to scrape"
    Krihelinator.Periodic.BigQuery.query()
    |> async_scrape()
    |> handle_scraped_data()
  end

  @doc """
  Get the rest of the repositories that weren't updated from github trending
  or BigQuery and rescrape.
  """
  def rescrape_still_dirty() do
    query = from(r in GH.Repo, where: r.dirty, select: r.name)
    names = Repo.all(query)
    Logger.info "Rescraping #{length(names)} still dirty repositories"
    names
    |> async_scrape
    |> handle_scraped_data()
  end

  @async_params [max_concurrency: 20, timeout: 60_000]

  @doc """
  Scrape repositories concurrently with `Task.async_stream`.
  """
  def async_scrape(names) do
    names
    |> Task.async_stream(&scrape_with_retries/1, @async_params)
    |> Stream.map(fn {:ok, cs} -> cs end)
  end

  @to_retry ~w(github_server_error timeout)a
  @max_attempts 3

  def scrape_with_retries(name, attempt \\ 1) do
    case Krihelinator.Scraper.scrape(name) do
      {:error, error} when error in @to_retry ->
        if attempt <= @max_attempts do<commit_msg>Disable scraping tasks timeout. The scraper already implements timeouts<commit_after>defmodule Krihelinator.Periodic do
  require Logger
  import Ecto.Query, only: [from: 2]
  alias Krihelinator.Repo
  alias Krihelinator.Github, as: GH

  @moduledoc """
  A background task that needs to run periodically on the Krihelinator server.

  - Mark all github repos as "dirty".
  - Scrape the github trending page for interesting, active, projects.
  - Using BigQuery, get and scrape all repos that at least 2 users pushed
    commits to.
  - Get the remaining "dirty" repos and pass through the scraper again, to
    update stats.
  - Clean the remaining dirty repos. These repos failed to update or fell bellow
    activity threshold.
  - Update the total krihelimeter for all languages.
  - Scrape showcases from github and update repos that belongs to showcases.
  """

  defp periodically_gc(pid) do
    # This is an ugly hack to force GC every now and then on the periodic
    # process. TODO search for a better solution!
    Process.sleep(30_000)
    if Process.alive?(pid) do
      :erlang.garbage_collect(pid)
      periodically_gc(pid)
    end
  end

  @doc """
  Main entry point.
  """
  def run() do
    my_pid = self()
    Task.async(fn -> periodically_gc(my_pid) end)
    Logger.info "Periodic process kicked in!"
    set_dirty_bit()
    scrape_trending()
    scrape_from_bigquery()
    rescrape_still_dirty()
    clean_dirty()
    update_languages_stats()
    scrape_showcases()
    Logger.info "Periodic process finished successfully!"
  end

  def set_dirty_bit() do
    Logger.info "Setting dirty bit for all"
    Repo.update_all(GH.Repo, set: [dirty: true])
  end

  @doc """
  Scrape the github trending page and update repos.
  """
  def scrape_trending() do
    Logger.info "Resetting trendiness for all"
    Repo.update_all(GH.Repo, set: [trending: false])
    Logger.info "Scraping trending"
    Krihelinator.Periodic.GithubTrending.scrape()
    |> async_scrape()
    |> Stream.map(fn
      {:ok, data} -> {:ok, Map.put(data, :trending, true)}
      otherwise -> otherwise
    end)
    |> handle_scraped_data()
  end

  @doc """
  Request repos from google BigQuery, scrape, and persist.
  """
  def scrape_from_bigquery() do
    Logger.info "Getting repositories from BigQuery to scrape"
    Krihelinator.Periodic.BigQuery.query()
    |> async_scrape()
    |> handle_scraped_data()
  end

  @doc """
  Get the rest of the repositories that weren't updated from github trending
  or BigQuery and rescrape.
  """
  def rescrape_still_dirty() do
    query = from(r in GH.Repo, where: r.dirty, select: r.name)
    names = Repo.all(query)
    Logger.info "Rescraping #{length(names)} still dirty repositories"
    names
    |> async_scrape
    |> handle_scraped_data()
  end

  @async_params [max_concurrency: 20, timeout: :infinity]

  @doc """
  Scrape repositories concurrently with `Task.async_stream`.
  """
  def async_scrape(names) do
    names
    |> Task.async_stream(&scrape_with_retries/1, @async_params)
    |> Stream.map(fn {:ok, cs} -> cs end)
  end

  @to_retry ~w(github_server_error timeout)a
  @max_attempts 3

  def scrape_with_retries(name, attempt \\ 1) do
    case Krihelinator.Scraper.scrape(name) do
      {:error, error} when error in @to_retry ->
        if attempt <= @max_attempts do
{'fraction_non_alphanumeric': 0.09464583649325042, 'fraction_numerical': 0.004398604580615805, 'mean_word_length': 3.349604221635884, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3839042', 'n_tokens_mistral': 2295, 'n_tokens_neox': 2108, 'n_words': 667}