.visidatarc

# Options
options.disp_column_sep = "│"
options.disp_keycol_sep = "║"
options.clean_names = True
options.disp_menu_fmt = ""
options.disp_sidebar = False
options.clipboard_copy_cmd = "pbcopy"
options.clipboard_paste_cmd = "pbpaste"
options.color_default = ""
options.quitguard = True
options.color_key_col = 110
options.color_note_type = 8
options.scroll_incr = -3
options.some_selected_rows = True
options.disp_histogram = "*"
options.undo = True
options.motd_url = ""
options.save_filetype = "jsonl"
options.motd_url=''

# Keybindings
vd.bindkey("0", "go-leftmost")
vd.bindkey("4", "go-rightmost")
vd.unbindkey("Up")
vd.unbindkey("Down")
vd.bindkey("Up", "setcol-precision-more")
vd.bindkey("Down", "setcol-precision-less")

# Toggle numeric binning
Sheet.addCommand("B", "toggle-binning", "options.numeric_binning = not options.numeric_binning")

import regex
import functools
import tldextract
import emoji
import urlexpander
import requests
import datetime
import fugashi
#import ucluster.vd.plugin
import deepl
from urllib.parse import urlparse
from ast import literal_eval
from nltk import word_tokenize
from nltk.util import ngrams
from nltk.corpus import stopwords
from dateutil import tz
from dateutil import parser
from geopy.geocoders import Nominatim
from snownlp import SnowNLP
from lingua import Language, LanguageDetectorBuilder
detector = LanguageDetectorBuilder.from_all_languages().with_preloaded_language_models().build()

translator = deepl.Translator("")

# In case you're running your own instance
geolocator = Nominatim(user_agent="SIO", domain="localhost:8080", scheme="http")
# geolocator = Nominatim(user_agent="SIO")


# Unshorten URLs with unshrtn. If the URL is shortened, return the best
# expanded URL we have, along with a bool saying it was a shortened URL.
# Otherwise, return the original, with bool and placeholders so we can compare
# 1:1 with links that weren't short to begin with.
@functools.lru_cache
def unshrtn(url):
    if urlexpander.is_short(url):
        resp = requests.get("http://localhost:3000", params={"url": url}).json()
        for key in ["canonical", "long", "short"]:
            if resp.get(key) is not None:
                return (resp[key], True, resp["status"], resp["title"])
    return (url, False, None, None)


# Do a cached geocode lookup via Nominatim
@functools.lru_cache
def geo(column):
    if column and column != "null":
        location = geolocator.geocode(column, timeout=5, language="en")
        if location is not None:
            latlong = str(location.latitude) + " " + str(location.longitude)
            return (latlong, location.address, location.address.split(",")[-1].strip())


# Convert to local time
def ltime(column):
    return parser.parse(column).astimezone(tz.tzlocal())


# Strip HTML tags
def nohtml(column):
    tag_re = regex.compile(r"(<!--.*?-->|<[^>]*>)")
    return tag_re.sub("", column)


# Convert abbreviated numbers like 1.1k to actual numbers
def unk(x):
    if x.isdigit():
        return int(x)
    else:
        if len(x) > 1:
            if x[-1] == "k":
                y = float(x[:-1]) * 1000
            elif x[-1] == "m":
                y = float(x[:-1]) * 1000000
    return int(y)


# Convert emoji to :this: representation, in memory of Chloe Price
def noemoji(column):
    if column:
        text = emoji.demojize(column)
        return text


# Extract all emoji into a list
def gemoji(column):
    emoji_list = []

    if column:
        data = regex.findall(r"\X", column)
        for word in data:
            if word in emoji.EMOJI_DATA:
                emoji_list.append(word)

    return emoji_list


# Extract all emoji into a text-based list
def temoji(column):
    emoji_list = gemoji(column)
    temoji_list = []
    for char in emoji_list:
        temoji = emoji.demojize(char)
        temoji_list.append(temoji)
    return temoji_list


def otherchars(column):
    if column:
        return [c for c in column if regex.match("[^\u0020-\u0370]", c)]


# Twitter helper functions (v1 or v2)
def hashtags(column):
    if column:
        if column.get("hashtags", []):
            try:
                return [hashtag["tag"] for hashtag in column["hashtags"]]

            except:
                return [hashtag["text"] for hashtag in column["hashtags"]]

        else:
            return None


def mentions(column):
    if column:
        if column.get("mentions", []):
            try:
                return [name["username"] for name in column["mentions"]]

            except:
                return [name["screen_name"] for name in column["user_mentions"]]

    else:
        return None


def annotations(column):
    if column:
        try:
            return [
                (anno["probability"], anno["type"], anno["normalized_text"])
                for anno in column["annotations"]
            ]

        except:
            return None


def urls(column):
    results = []
    try:
        for url in column["urls"]:
            if "unwound_url" in url:
                results.append(url["unwound_url"])
            elif url.get("expanded_url"):
                results.append(url["expanded_url"])
            elif url.get("url"):
                results.append(url["url"])

        return results

    except:
        return None


def domains(column):
    if column:
        if type(column) is list:
            doms = []
            for elem in column:
                doms.append(tldextract.extract(elem).registered_domain)
            return doms
        else:
            return tldextract.extract(column).registered_domain


# Sometimes we get data in a CSV or something that's a literal string,
# but with array notation.
def strtoarray(column):
    if column:
        return literal_eval(column)


# These re- functions are for plain text fields where we don't have
# these as separate metadata/columns
def rementions(column):
    if column:
        return regex.findall(r"\B@\w\w+", column)


def rehashtags(column):
    if column:
        return regex.findall(r"\B#\w\w+", column)

def loras(column):
    if column:
        return regex.findall(r"(<[Ll]ora:.*:???>)", column)

def reurls(column):
    r = r"(?i)\b((?:https?://|www\d{0,3}[.]|[a-z0-9.\-]+[.][a-z]{2,4}/)(?:[^\s()<>]+|\(([^\s()<>]+|(\([^\s()<>]+\)))*\))+(?:\(([^\s()<>]+|(\([^\s()<>]+\)))*\)|[^\s`!()\[\]{};:'\".,<>?«»“”‘’]))"
    if column:
        url = regex.findall(r, column)
        return [x[0] for x in url]
    else:
        return ""


# Normalize URLs for cleaner freq analysis
def normurl(column):
    if column:
        url = urlparse(column)[1:]
        url = "".join(url)
        if url.endswith("/"):
            url = url[:-1]
        if url.startswith("www."):
            url = url[4:]

        return url


# Detect language and probability
@functools.lru_cache
def ld(column):
    if column and len(column) >= 5:
        confidence_values = detector.compute_language_confidence_values(column)
        return (confidence_values[0].language.iso_code_639_1.name, confidence_values[0].value)
    return ""


@functools.lru_cache
def translate(column):
    if column and len(column) >= 10:
        return str(translator.translate_text(column, target_lang="EN-US"))
    return ""

# read in stubs.txt from tmp. If the stub matches the first part of string passed in as column followed by a capital letter, return True.

def stubs(column):
    if column:
        with open("/tmp/stubs.txt") as f:
            stubs = f.read().splitlines()
            for stub in stubs:
                if column.startswith(stub) and column[len(stub)].isupper():
                    return True
    return False

# Sloppy ngram filter
def grams(column, lang, n):
    if column:
        stop = stopwords.words(lang)
        stop.extend(["http", "https", "RT"])
        w = word_tokenize(column)
        alnum = [word for word in w if word.isalnum()]
        clean = [word for word in alnum if not word in stop]
        tokens = ngrams(clean, n)
        return [" ".join(thegrams) for thegrams in tokens]


# Chinese language helper to pull out words
@functools.lru_cache
def cnwords(column):
    if column:
        s = SnowNLP(column)
        return s.words


# or sentences
@functools.lru_cache
def cnsent(column):
    if column:
        s = SnowNLP(column)
        return s.sentences


# or keywords
@functools.lru_cache
def cnkw(column, count):
    if column:
        s = SnowNLP(column)
        return s.keywords(count)


# And now for jp
tagger = fugashi.Tagger()
@functools.lru_cache
def jpgrams(column):
    if column:
        s = [word.surface for word in tagger(column)]
        return s


# vim: syntax=python