<commit_before># profile.py

from translationstring import TranslationString as _

from error import WeasylError
import macro as m
import define as d

import pytz

import orm
import shout
import welcome

from libweasyl.html import strip_html
from libweasyl.models import tables
from libweasyl import ratings
from libweasyl import staff

from weasyl.cache import region
from weasyl.configuration_builder import create_configuration, BoolOption, ConfigOption
from weasyl import media


class ExchangeType:
    def __init__(self, name_singular, name_plural):
        self.name_singular = name_singular
        self.name_plural = name_plural

EXCHANGE_TYPE_TRADE = ExchangeType("trade", "trades")
EXCHANGE_TYPE_REQUEST = ExchangeType("request", "requests")
EXCHANGE_TYPE_COMMISSION = ExchangeType("commission", "commissions")


class ExchangeSetting:
    def __init__(self, code, text):
        self.code = code
        self.text = text

    def format(self, request_type):
        return _(self.text.format(type=request_type))


EXCHANGE_SETTING_ACCEPTING = ExchangeSetting("o", "I am currently accepting {type.name_plural}")
EXCHANGE_SETTING_SOMETIMES = ExchangeSetting("s", "I may sometimes accept {type.name_plural}")
EXCHANGE_SETTING_FULL_QUEUE = ExchangeSetting("f", "My {type.name_singular} queue is currently filled")
EXCHANGE_SETTING_NOT_ACCEPTING = ExchangeSetting("c", "I am not accepting {type.name_plural} right now")
EXCHANGE_SETTING_NOT_APPLICABLE = ExchangeSetting("e", "This is not applicable to me")

ALL_EXCHANGE_SETTINGS = [EXCHANGE_SETTING_ACCEPTING, EXCHANGE_SETTING_SOMETIMES,
                         EXCHANGE_SETTING_FULL_QUEUE, EXCHANGE_SETTING_NOT_ACCEPTING,
                         EXCHANGE_SETTING_NOT_APPLICABLE]
EXCHANGE_SETTING_CODE_MAP = {setting.code: setting for setting in ALL_EXCHANGE_SETTINGS}

ALLOWABLE_EXCHANGE_CODES = {
    EXCHANGE_TYPE_TRADE: 'osce',
    EXCHANGE_TYPE_REQUEST: 'osce',
    EXCHANGE_TYPE_COMMISSION: 'osfce',
}


Config = create_configuration([
    BoolOption("twelvehour", "2"),
    ConfigOption("rating", dict(zip(ratings.ALL_RATINGS, ["", "m", "a", "p"]))),
    BoolOption("tagging", "k"),
    BoolOption("edittagging", "r"),
    BoolOption("hideprofile", "h"),
    BoolOption("hidestats", "i"),
    BoolOption("hidefavorites", "v"),
    BoolOption("hidefavbar", "u"),
    ConfigOption("shouts", {"anyone": "", "friends_only": "x", "staff_only": "w"}),
    ConfigOption("notes", {"anyone": "", "friends_only": "z", "staff_only": "y"}),
    BoolOption("filter", "l"),
    BoolOption("follow_s", "s"),
    BoolOption("follow_c", "c"),
    BoolOption("follow_f", "f"),
    BoolOption("follow_t", "t"),
    BoolOption("follow_j", "j"),
])


def get_exchange_setting(exchange_type, code):
    if code not in ALLOWABLE_EXCHANGE_CODES[exchange_type]:
        return EXCHANGE_SETTING_NOT_ACCEPTING
    return EXCHANGE_SETTING_CODE_MAP[code]


def resolve(userid, otherid, othername, myself=True):
    """
    Attempts to determine the userid of a specified user; resolves using otherid,
    othername, and userid (if myself is True), in that order. If no userid can be
    resolved, returns 0 instead.
    """
    result = None

    if otherid:
        result = d.execute("SELECT userid FROM login WHERE userid = %i", [d.get_int(otherid)], ["element"])

        if result:
            return result
    elif othername:
        result = d.execute("SELECT userid FROM login WHERE login_name = '%s'", [d.get_sysname(othername)], ["element"])

        if result:
            return result

        result = d.execute("SELECT userid FROM useralias WHERE alias_name = '%s'", [d.get_sysname(othername)], ["element"])

        if result:
            return result
    elif userid and myself:
        return userid

    return 0


@region.cache_on_arguments()
@d.record_timing
def resolve_by_login(login):
    return resolve(None, None, login, False)


def select_profile(userid, avatar=False, banner=False, propic=False, images=False, commish=True, viewer=None):
    query = d.execute("""
        SELECT pr.username, pr.full_name, pr.catchphrase, pr.unixtime, pr.profile_text,
            pr.settings, pr.stream_url, pr.config, pr.stream_text, lo.settings, us.end_time
        FROM profile pr
            INNER JOIN login lo USING (userid)
            LEFT JOIN user_streams us USING (userid)
        WHERE userid = %i
    """, [userid], ["single"])

    if not query:
        raise WeasylError('RecordMissing')

    streaming_status = "stopped"
    if query[6]:  # profile.stream_url
        if query[10] > d.get_time():  # user_streams.end_time
            streaming_status = "started"
        elif 'l' in query[5]:
            streaming_status = "later"

    return {
        "userid": userid,
        "user_media": media.get_user_media(userid),
        "username": query[0],
        "full_name": query[1],
        "catchphrase": query[2],
        "unixtime": query[3],
        "profile_text": query[4],
        "settings": query[5],
        "stream_url": query[6],
        "stream_text": query[8],
        "config": query[7],
        "show_favorites_bar": "u" not in query[7] and "v" not in query[7],
        "show_favorites_tab": userid == viewer or "v" not in query[7],
        "commish_slots": 0,
        "banned": "b" in query[9],
        "suspended": "s" in query[9],
        "streaming_status": streaming_status,
    }


def twitter_card(userid):
    username, full_name, catchphrase, profile_text, config, twitter = d.execute(
        "SELECT pr.username, pr.full_name, pr.catchphrase, pr.profile_text, pr.config, ul.link_value "
        "FROM profile pr "
        "LEFT JOIN user_links ul ON pr.userid = ul.userid AND ul.link_type = 'twitter' "
        "WHERE pr.userid = %i",
        [userid],
        ["single"])

    ret = {
        'card': 'summary',
        'url': d.absolutify_url('/~%s' % (username,)),
        'title': '%s on Weasyl' % (full_name,),
    }

    if catchphrase:
        description = '"%s"' % (catchphrase,)
    elif profile_text:
        description = strip_html(profile_text)
    else:
        description = "[%s has an empty profile, but is eggcelent!]" % (full_name,)
    ret['description'] = d.summarize(description)

    media_items = media.get_user_media(userid)
    ret['image:src'] = d.absolutify_url(media_items['avatar'][0]['display_url'])

    if twitter:
        ret['creator'] = '@%s' % (twitter.lstrip('@'),)

    return ret


def select_myself(userid):
    if not userid:
        return

    query = d.execute("SELECT username, config FROM profile WHERE userid = %i", [userid], ["single"])

    return {
        "userid": userid,
        "username": query[0],
        "is_mod": userid in staff.MODS,
        "user_media": media.get_user_media(userid),
    }


def get_user_age(userid):
    assert userid
    return d.convert_age(d.execute("SELECT birthday FROM userinfo WHERE userid = %i", [userid], ["element"]))


def get_user_ratings(userid):
    return ratings.get_ratings_for_age(get_user_age(userid))


def check_user_rating_allowed(userid, rating):
    # TODO(kailys): ensure usages always pass a Rating
    minimum_age = rating.minimum_age if isinstance(rating, ratings.Rating) else ratings.CODE_MAP[rating].minimum_age
    if get_user_age(userid) < minimum_age:
        raise WeasylError("ratingInvalid")


def select_userinfo(userid, config=None):
    if config is None:
        [query] = d.engine.execute("""
            SELECT pr.config, ui.birthday, ui.gender, ui.country
            FROM profile pr
            INNER JOIN userinfo ui USING (userid)
            WHERE pr.userid = %(userid)s
        """, userid=userid)
    else:
        [query] = d.engine.execute("""
            SELECT %(config)s, birthday, gender, country
            FROM userinfo
            WHERE userid = %(userid)s
        """, userid=userid, config=config)

    user_link_rows = d.engine.execute("""
        SELECT link_type, ARRAY_AGG(link_value)
        FROM user_links
        WHERE userid = %(userid)s
        GROUP BY link_type
    """, userid=userid)
    user_links = {r[0]: r[1] for r in user_link_rows}

    show_age = "b" in query[0] or d.get_userid() in staff.MODS
    return {
        "birthday": query[1],
        "age": d.convert_age(query[1]) if show_age else None,
        "show_age": "b" in query[0],
        "gender": query[2],
        "country": query[3],
        "user_links": user_links,
    }


def select_report_stats(userid):
    query_byreason = d.engine.execute("""
        SELECT count(r.reportid), r.closure_reason FROM report r
        JOIN reportcomment c ON r.reportid = c.reportid
        WHERE c.userid = %(userid)s AND r.closed_at IS NOT null
        GROUP BY r.closure_reason
        UNION
        SELECT count(r.reportid), r.closure_reason FROM report r
        JOIN reportcomment c ON r.reportid = c.reportid
        WHERE c.userid = %(userid)s AND r.closed_at IS null
        GROUP BY r.closure_reason
    """, userid=userid)
    # create a dict of {'closure_reason' : 'count'}
    # closure_reason will be None if report was not yet closed.
    return {row[1].replace("-", " ").title() if row[1] is not None else "Open":
            row[0] for row in query_byreason}


def select_relation(userid, otherid):
    if not userid or userid == otherid:
        return {
            "follow": False,
            "friend": False,
            "ignore": False,
            "friendreq": False,
            "is_self": userid == otherid,
        }

    query = d.engine.execute("""
        SELECT
            (SELECT EXISTS (SELECT 0 FROM watchuser WHERE (userid, otherid) = (%(user)s, %(other)s))),
            (SELECT EXISTS (SELECT 0 FROM frienduser WHERE userid IN (%(user)s, %(other)s) AND otherid IN (%(user)s, %(other)s) AND settings !~ 'p')),
            (SELECT EXISTS (SELECT 0 FROM ignoreuser WHERE (userid, otherid) = (%(user)s, %(other)s))),
            (SELECT EXISTS (SELECT 0 FROM frienduser WHERE (userid, otherid) = (%(user)s, %(other)s) AND settings ~ 'p')),
            (SELECT EXISTS (SELECT 0 FROM watchuser WHERE (userid, otherid) = (%(other)s, %(user)s)))
    """, user=userid, other=otherid).first()

    return {
        "follow": query[0],
        "friend": query[1],
        "ignore": query[2],
        "friendreq": query[3],
        "follower": query[4],
        "is_self": False,
    }


@region.cache_on_arguments(expiration_time=600)
@d.record_timing
def _select_statistics(userid):
    query = d.execute("""
        SELECT<commit_msg>changed the query<commit_after># profile.py

from translationstring import TranslationString as _

from error import WeasylError
import macro as m
import define as d

import pytz

import orm
import shout
import welcome

from libweasyl.html import strip_html
from libweasyl.models import tables
from libweasyl import ratings
from libweasyl import staff

from weasyl.cache import region
from weasyl.configuration_builder import create_configuration, BoolOption, ConfigOption
from weasyl import media


class ExchangeType:
    def __init__(self, name_singular, name_plural):
        self.name_singular = name_singular
        self.name_plural = name_plural

EXCHANGE_TYPE_TRADE = ExchangeType("trade", "trades")
EXCHANGE_TYPE_REQUEST = ExchangeType("request", "requests")
EXCHANGE_TYPE_COMMISSION = ExchangeType("commission", "commissions")


class ExchangeSetting:
    def __init__(self, code, text):
        self.code = code
        self.text = text

    def format(self, request_type):
        return _(self.text.format(type=request_type))


EXCHANGE_SETTING_ACCEPTING = ExchangeSetting("o", "I am currently accepting {type.name_plural}")
EXCHANGE_SETTING_SOMETIMES = ExchangeSetting("s", "I may sometimes accept {type.name_plural}")
EXCHANGE_SETTING_FULL_QUEUE = ExchangeSetting("f", "My {type.name_singular} queue is currently filled")
EXCHANGE_SETTING_NOT_ACCEPTING = ExchangeSetting("c", "I am not accepting {type.name_plural} right now")
EXCHANGE_SETTING_NOT_APPLICABLE = ExchangeSetting("e", "This is not applicable to me")

ALL_EXCHANGE_SETTINGS = [EXCHANGE_SETTING_ACCEPTING, EXCHANGE_SETTING_SOMETIMES,
                         EXCHANGE_SETTING_FULL_QUEUE, EXCHANGE_SETTING_NOT_ACCEPTING,
                         EXCHANGE_SETTING_NOT_APPLICABLE]
EXCHANGE_SETTING_CODE_MAP = {setting.code: setting for setting in ALL_EXCHANGE_SETTINGS}

ALLOWABLE_EXCHANGE_CODES = {
    EXCHANGE_TYPE_TRADE: 'osce',
    EXCHANGE_TYPE_REQUEST: 'osce',
    EXCHANGE_TYPE_COMMISSION: 'osfce',
}


Config = create_configuration([
    BoolOption("twelvehour", "2"),
    ConfigOption("rating", dict(zip(ratings.ALL_RATINGS, ["", "m", "a", "p"]))),
    BoolOption("tagging", "k"),
    BoolOption("edittagging", "r"),
    BoolOption("hideprofile", "h"),
    BoolOption("hidestats", "i"),
    BoolOption("hidefavorites", "v"),
    BoolOption("hidefavbar", "u"),
    ConfigOption("shouts", {"anyone": "", "friends_only": "x", "staff_only": "w"}),
    ConfigOption("notes", {"anyone": "", "friends_only": "z", "staff_only": "y"}),
    BoolOption("filter", "l"),
    BoolOption("follow_s", "s"),
    BoolOption("follow_c", "c"),
    BoolOption("follow_f", "f"),
    BoolOption("follow_t", "t"),
    BoolOption("follow_j", "j"),
])


def get_exchange_setting(exchange_type, code):
    if code not in ALLOWABLE_EXCHANGE_CODES[exchange_type]:
        return EXCHANGE_SETTING_NOT_ACCEPTING
    return EXCHANGE_SETTING_CODE_MAP[code]


def resolve(userid, otherid, othername, myself=True):
    """
    Attempts to determine the userid of a specified user; resolves using otherid,
    othername, and userid (if myself is True), in that order. If no userid can be
    resolved, returns 0 instead.
    """
    result = None

    if otherid:
        result = d.execute("SELECT userid FROM login WHERE userid = %i", [d.get_int(otherid)], ["element"])

        if result:
            return result
    elif othername:
        result = d.execute("SELECT userid FROM login WHERE login_name = '%s'", [d.get_sysname(othername)], ["element"])

        if result:
            return result

        result = d.execute("SELECT userid FROM useralias WHERE alias_name = '%s'", [d.get_sysname(othername)], ["element"])

        if result:
            return result
    elif userid and myself:
        return userid

    return 0


@region.cache_on_arguments()
@d.record_timing
def resolve_by_login(login):
    return resolve(None, None, login, False)


def select_profile(userid, avatar=False, banner=False, propic=False, images=False, commish=True, viewer=None):
    query = d.execute("""
        SELECT pr.username, pr.full_name, pr.catchphrase, pr.unixtime, pr.profile_text,
            pr.settings, pr.stream_url, pr.config, pr.stream_text, lo.settings, us.end_time
        FROM profile pr
            INNER JOIN login lo USING (userid)
            LEFT JOIN user_streams us USING (userid)
        WHERE userid = %i
    """, [userid], ["single"])

    if not query:
        raise WeasylError('RecordMissing')

    streaming_status = "stopped"
    if query[6]:  # profile.stream_url
        if query[10] > d.get_time():  # user_streams.end_time
            streaming_status = "started"
        elif 'l' in query[5]:
            streaming_status = "later"

    return {
        "userid": userid,
        "user_media": media.get_user_media(userid),
        "username": query[0],
        "full_name": query[1],
        "catchphrase": query[2],
        "unixtime": query[3],
        "profile_text": query[4],
        "settings": query[5],
        "stream_url": query[6],
        "stream_text": query[8],
        "config": query[7],
        "show_favorites_bar": "u" not in query[7] and "v" not in query[7],
        "show_favorites_tab": userid == viewer or "v" not in query[7],
        "commish_slots": 0,
        "banned": "b" in query[9],
        "suspended": "s" in query[9],
        "streaming_status": streaming_status,
    }


def twitter_card(userid):
    username, full_name, catchphrase, profile_text, config, twitter = d.execute(
        "SELECT pr.username, pr.full_name, pr.catchphrase, pr.profile_text, pr.config, ul.link_value "
        "FROM profile pr "
        "LEFT JOIN user_links ul ON pr.userid = ul.userid AND ul.link_type = 'twitter' "
        "WHERE pr.userid = %i",
        [userid],
        ["single"])

    ret = {
        'card': 'summary',
        'url': d.absolutify_url('/~%s' % (username,)),
        'title': '%s on Weasyl' % (full_name,),
    }

    if catchphrase:
        description = '"%s"' % (catchphrase,)
    elif profile_text:
        description = strip_html(profile_text)
    else:
        description = "[%s has an empty profile, but is eggcelent!]" % (full_name,)
    ret['description'] = d.summarize(description)

    media_items = media.get_user_media(userid)
    ret['image:src'] = d.absolutify_url(media_items['avatar'][0]['display_url'])

    if twitter:
        ret['creator'] = '@%s' % (twitter.lstrip('@'),)

    return ret


def select_myself(userid):
    if not userid:
        return

    query = d.execute("SELECT username, config FROM profile WHERE userid = %i", [userid], ["single"])

    return {
        "userid": userid,
        "username": query[0],
        "is_mod": userid in staff.MODS,
        "user_media": media.get_user_media(userid),
    }


def get_user_age(userid):
    assert userid
    return d.convert_age(d.execute("SELECT birthday FROM userinfo WHERE userid = %i", [userid], ["element"]))


def get_user_ratings(userid):
    return ratings.get_ratings_for_age(get_user_age(userid))


def check_user_rating_allowed(userid, rating):
    # TODO(kailys): ensure usages always pass a Rating
    minimum_age = rating.minimum_age if isinstance(rating, ratings.Rating) else ratings.CODE_MAP[rating].minimum_age
    if get_user_age(userid) < minimum_age:
        raise WeasylError("ratingInvalid")


def select_userinfo(userid, config=None):
    if config is None:
        [query] = d.engine.execute("""
            SELECT pr.config, ui.birthday, ui.gender, ui.country
            FROM profile pr
            INNER JOIN userinfo ui USING (userid)
            WHERE pr.userid = %(userid)s
        """, userid=userid)
    else:
        [query] = d.engine.execute("""
            SELECT %(config)s, birthday, gender, country
            FROM userinfo
            WHERE userid = %(userid)s
        """, userid=userid, config=config)

    user_link_rows = d.engine.execute("""
        SELECT link_type, ARRAY_AGG(link_value)
        FROM user_links
        WHERE userid = %(userid)s
        GROUP BY link_type
    """, userid=userid)
    user_links = {r[0]: r[1] for r in user_link_rows}

    show_age = "b" in query[0] or d.get_userid() in staff.MODS
    return {
        "birthday": query[1],
        "age": d.convert_age(query[1]) if show_age else None,
        "show_age": "b" in query[0],
        "gender": query[2],
        "country": query[3],
        "user_links": user_links,
    }


def select_report_stats(userid):
    query_byreason = d.engine.execute("""
        SELECT count(r.reportid), r.closure_reason FROM report r
        JOIN reportcomment c ON r.reportid = c.reportid
        WHERE c.userid = %(userid)s AND r.closed_at IS NOT null
        GROUP BY r.closure_reason
        UNION
        SELECT count(r.reportid), r.closure_reason FROM report r
        JOIN reportcomment c ON r.reportid = c.reportid
        WHERE c.userid = %(userid)s AND r.closed_at IS null
        GROUP BY r.closure_reason
    """, userid=userid)
    # create a dict of {'closure_reason' : 'count'}
    # closure_reason will be None if report was not yet closed.
    return {row[1].replace("-", " ").title() if row[1] is not None else "Open":
            row[0] for row in query_byreason}


def select_relation(userid, otherid):
    if not userid or userid == otherid:
        return {
            "follow": False,
            "friend": False,
            "ignore": False,
            "friendreq": False,
            "is_self": userid == otherid,
        }

    query = d.engine.execute("""
        SELECT
            (SELECT EXISTS (SELECT 0 FROM watchuser WHERE (userid, otherid) = (%(user)s, %(other)s)) AS follow),
            (SELECT EXISTS (SELECT 0 FROM frienduser WHERE userid IN (%(user)s, %(other)s) AND otherid IN (%(user)s, %(other)s) AND settings !~ 'p') AS friend),
            (SELECT EXISTS (SELECT 0 FROM ignoreuser WHERE (userid, otherid) = (%(user)s, %(other)s)) AS ignore),
            (SELECT EXISTS (SELECT 0 FROM frienduser WHERE (userid, otherid) = (%(user)s, %(other)s) AND settings ~ 'p') AS friendreq),
            (SELECT EXISTS (SELECT 0 FROM watchuser WHERE (userid, otherid) = (%(other)s, %(user)s)) AS follower)
    """, user=userid, other=otherid).first()

    return dict(
        query,
        is_self=False)


@region.cache_on_arguments(expiration_time=600)
@d.record_timing
def _select_statistics(userid):
    query = d.execute("""
        SELECT
{'fraction_non_alphanumeric': 0.12628149851489892, 'fraction_numerical': 0.004359490274983233, 'mean_word_length': 3.143509329098849, 'pattern_counts': {'":': 84, '<': 5, '<?xml version=': 0, '>': 5, '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': '7274919', 'n_tokens_mistral': 6870, 'n_tokens_neox': 6722, 'n_words': 1712}