import re
from collections import Counter
import numpy as np
import pandas as pd
from .utils import smooth, sort, softmax
[docs]
class SentimentLexicon:
labels = [] # overridden by concrete subclasses
_feature = None # bundled-data key (see kosac/data/), set by subclasses
def __init__(self, filepath=None, ngrams=[1]):
self.ngrams = ngrams
self.min_freq = 0
self.threshold = 0.0
if filepath:
# keep_default_na=False so the literal 'None' label (a valid polarity /
# intensity value) stays a string, not a missing value.
df = pd.read_csv(filepath, keep_default_na=False)
# Concrete subclasses declare `labels`; a GenericLexicon infers them from
# the CSV's count columns (everything but `ngram`), enabling round-trips.
if not self.labels:
self.labels = [c for c in df.columns if c != 'ngram']
df['entry'] = df['ngram']
# The CSV stores only absolute label counts; derive freq (= the number of
# Seeds = the row total) and max.value / max.prop.
if self.labels:
counts = df[self.labels].astype(int)
df['freq'] = counts.sum(axis=1)
df['max.value'] = counts.idxmax(axis=1)
df['max.prop'] = counts.max(axis=1) / counts.sum(axis=1)
df = df.sort_values('max.prop', ascending=False)
df['ngram'] = df['entry'].str.count(' ') + 1
df = df[df['ngram'].isin(self.ngrams)]
df = df.sort_values('ngram', ascending=False)
df.sort_values('entry', inplace=True)
df.set_index('entry', inplace=True)
else:
df = pd.DataFrame()
df.index.name = 'entry'
self.original_lexicon = df
self.lexicon = self.original_lexicon.copy()
[docs]
@classmethod
def load(cls, ngrams=[1], min_freq=0, threshold=0.0):
"""Load this feature's lexicon from the CSV bundled with the package."""
if cls._feature is None:
raise TypeError(
f'{cls.__name__} has no bundled data; use a concrete subclass such as '
'PolarityLexicon, or pass an explicit filepath to the constructor.'
)
from ._resources import resource_path
with resource_path(cls._feature) as path:
lexicon = cls(filepath=str(path), ngrams=ngrams)
if min_freq or threshold:
lexicon.set_lexicon(min_freq=min_freq, threshold=threshold)
return lexicon
def __repr__(self):
name = type(self).__name__
return f'{name}(ngrams={self.ngrams}, min_freq={self.min_freq}, threshold={self.threshold})'
def __eq__(self, other):
self.lexicon == other.lexicon
def __ne__(self, other):
self.lexicon != other.lexicon
def __add__(self, other):
raise NotImplementedError
[docs]
def get_original_lexicon(self):
return self.original_lexicon
[docs]
def save(self, filepath):
"""Write the lexicon to a CSV in the package's absolute-count format.
Columns are `ngram` (morphemes joined by a space) and one absolute-count
column per label — the same format the loader reads. Reload it through the
constructor: a concrete subclass uses its declared labels, while
`GenericLexicon(filepath=...)` infers the labels from the columns.
"""
out = self.lexicon[self.labels].astype(int).copy()
out.insert(0, 'ngram', out.index)
out.to_csv(filepath, index=False)
return filepath
[docs]
def set_lexicon(self, min_freq=0, threshold=0.0):
# Re-applicable filter: always start from the originally loaded lexicon so the
# threshold can be loosened as well as tightened. Falls back to the current
# lexicon for from-scratch builds (where original_lexicon is empty).
# TODO: frequency 대신 tf-idf
df = self.original_lexicon if len(self.original_lexicon) else self.lexicon
self.lexicon = df[(df['freq'] >= min_freq) & (df['max.prop'] > threshold)]
self.min_freq = min_freq
self.threshold = threshold
[docs]
def get_lexicon(self):
return self.lexicon
[docs]
def get_size(self):
return len(self.lexicon)
[docs]
def get_labels(self):
return self.labels
[docs]
def get_entry(self, morph):
return self.lexicon.loc[morph]
[docs]
def verify(self, morph, verbose=True):
counts = self.lexicon.loc[morph, self.labels].astype('int')
self.lexicon.loc[morph, 'freq'] = counts.sum()
self.lexicon.loc[morph, 'max.value'] = counts.idxmax()
self.lexicon.loc[morph, 'max.prop'] = counts.max() / counts.sum()
if verbose:
print(self.lexicon.loc[morph])
[docs]
def initialize_entry(self, morph, **kwargs):
row = pd.Series(dtype='object')
row['ngram'] = morph.count(' ') + 1
row['freq'] = 0
for label in self.labels:
row[label] = 0
counts = row[self.labels]
row['freq'] = counts.sum()
row['max.value'] = counts.idxmax()
row['max.prop'] = 0.
if len(self.lexicon.columns) == 0:
# First insert into an empty lexicon: establish the columns first.
self.lexicon = pd.DataFrame(columns=row.index)
self.lexicon.index.name = 'entry'
self.lexicon.loc[morph] = row
[docs]
def add_token(self, morph, tag, verbose=True):
if morph not in self.lexicon.index:
self.initialize_entry(morph)
self.lexicon.loc[morph, tag] += 1
self.verify(morph, verbose)
[docs]
def update(self, examples):
for (morph, tag) in examples:
self.add_token(morph, tag, verbose=False)
[docs]
def update_from_corpus(self, corpus, tokenizer, pos_tag=None, min_freq=0,
max_value_threshold=0.0):
"""Rebuild the lexicon's frequencies from a labeled corpus.
Each text is tokenized into morpheme N-grams and its label is counted for
every N-gram. The optional filters keep the result from being dominated by
uninformative entries:
Parameters
----------
pos_tag : str or iterable of str, optional
Keep only N-grams that contain at least one token with one of these POS
tags (e.g. ``{'NNG', 'NNP', 'VV', 'VA', 'XR', 'MAG'}`` for content words).
This drops pure function-morpheme entries (``이/MM``, ``다/EF``) while
keeping meaningful constructions like ``ㄹ/ETM 수/NNB 있/VV`` ("can ...").
min_freq : int
Drop entries observed fewer than ``min_freq`` times.
max_value_threshold : float
Drop entries whose dominant-label proportion (``max.prop``) is below this.
Notes
-----
This is a full rebuild from the corpus: any counts previously in the lexicon
are discarded and the result holds exactly the N-grams observed in ``corpus``.
Counts are tallied with a :class:`~collections.Counter` and assembled in one
vectorized pass, so a 150k-sentence corpus builds in seconds rather than the
minutes a per-token pandas update would take.
"""
texts = corpus.df['text'].astype('str').tolist()
corpus.df['entry'] = tokenizer.get_ngrams_batch(texts, self.ngrams)
exploded = corpus.df[['entry', 'label']].explode('entry')
pairs = ((entry, label) for entry, label in zip(exploded['entry'], exploded['label'])
if isinstance(entry, str) and entry)
if pos_tag is not None:
allowed = {pos_tag} if isinstance(pos_tag, str) else set(pos_tag)
pairs = ((entry, label) for (entry, label) in pairs
if any(token.rsplit('/', 1)[-1] in allowed for token in entry.split(' ')))
self.lexicon = self._lexicon_from_tally(Counter(pairs))
if min_freq or max_value_threshold:
df = self.lexicon
keep = (df['freq'] >= min_freq) & (df['max.prop'] >= max_value_threshold)
self.lexicon = df[keep]
def _lexicon_from_tally(self, tally):
"""Assemble the lexicon DataFrame from a ``{(entry, label): count}`` tally.
Builds the absolute-count matrix in one shot and derives ``freq`` /
``max.value`` / ``max.prop`` the same way the CSV loader does. Labels outside
``self.labels`` are an error (the corpus must use this lexicon's labels).
"""
columns = ['ngram', 'freq', *self.labels, 'max.value', 'max.prop']
unknown = {label for (_entry, label) in tally} - set(self.labels)
if unknown:
raise ValueError(
f'corpus labels {sorted(unknown)} are not in this lexicon\'s labels '
f'{self.labels}; set matching labels (e.g. GenericLexicon.set_labels).')
entries = sorted({entry for (entry, _label) in tally})
if not entries:
df = pd.DataFrame(columns=columns)
df.index.name = 'entry'
return df
label_col = {label: i for i, label in enumerate(self.labels)}
row_of = {entry: i for i, entry in enumerate(entries)}
matrix = np.zeros((len(entries), len(self.labels)), dtype=int)
for (entry, label), n in tally.items():
matrix[row_of[entry], label_col[label]] = n
counts = pd.DataFrame(matrix, index=pd.Index(entries, name='entry'),
columns=self.labels)
freq = counts.sum(axis=1)
df = pd.DataFrame(index=counts.index)
df['ngram'] = [entry.count(' ') + 1 for entry in entries]
df['freq'] = freq
for label in self.labels:
df[label] = counts[label]
df['max.value'] = counts.idxmax(axis=1)
df['max.prop'] = counts.max(axis=1) / freq
return df[columns]
[docs]
def export_user_dict(self, dict_path='user_dictionary.txt'):
unigrams = self.lexicon[self.lexicon['ngram'] == 1].index.tolist()
with open(dict_path, 'w') as f:
f.writelines('\n'.join(['\t'.join(unigram.split('/')) for unigram in unigrams]))
print('USER_DICT PATH:', dict_path)
self.dict_path = dict_path
[docs]
def get_pattern(self, sorting=True):
my_lexicon = self.lexicon.copy()
if sorting:
sorts = sort(my_lexicon)
else:
sorts = my_lexicon
# Entries are matched literally: some contain regex-special characters
# (e.g. the wildcard '*' in '가*/JKS'), so each entry must be escaped.
return re.compile('|'.join(re.escape(entry) for entry in sorts.index))
[docs]
def match_patterns(self, sentence, tokenizer, sorting=True):
pattern = self.get_pattern(sorting)
tagged = tokenizer.get_tokens_str(sentence)
matches = pattern.findall(tagged)
return matches
[docs]
def get_match_info(self, sentence, tokenizer, sorting=True):
matches = self.match_patterns(sentence, tokenizer, sorting)
result = [(match, self.lexicon.loc[match, 'max.value'], self.lexicon.loc[match, 'max.prop']) for match in matches]
return result
[docs]
def get_smoothed_lexicon(self):
return self.lexicon.apply(smooth, labels=self.labels, axis=1)
[docs]
def get_sent_probs(self, sentence, tokenizer, smoothing=True):
matches = self.match_patterns(sentence, tokenizer)
if not matches:
# No lexicon entry matched: no evidence, so return a uniform distribution
# rather than feeding an empty frame to softmax.
return pd.Series(1 / len(self.labels), index=self.labels)
frequencies = self.lexicon.loc[matches].copy()
if smoothing:
smoothed = frequencies.apply(smooth, labels=self.labels, axis=1)
else:
smoothed = frequencies[self.labels]
return softmax(np.log(smoothed).sum()).sort_values(ascending=False)
# Legacy aliases for the original SentLex notebook API.
get = get_entry
match = match_patterns
[docs]
class PolarityLexicon(SentimentLexicon):
"""Sentiment polarity: POS / NEG / NEUT / COMP (mixed) / None."""
labels = ['COMP', 'NEG', 'NEUT', 'None', 'POS']
_feature = 'polarity'
[docs]
class IntensityLexicon(SentimentLexicon):
"""Sentiment intensity: High / Medium / Low / None."""
labels = ['High', 'Low', 'Medium', 'None']
_feature = 'intensity'
[docs]
class ExpressiveTypeLexicon(SentimentLexicon):
"""How the sentiment is expressed (direct/indirect/writing-device, ...)."""
labels = ['dir-action', 'dir-explicit', 'dir-speech', 'indirect', 'writing-device']
_feature = 'expressive-type'
[docs]
class NestedOrderLexicon(SentimentLexicon):
"""Nesting depth of the subjective expression (0–3)."""
labels = ['0', '1', '2', '3']
_feature = 'nested-order'
[docs]
class SubjectivityPolarityLexicon(SentimentLexicon):
"""Polarity of the subjectivity: POS / NEG / NEUT / COMP."""
labels = ['COMP', 'NEG', 'NEUT', 'POS']
_feature = 'subjectivity-polarity'
[docs]
class SubjectivityTypeLexicon(SentimentLexicon):
"""Type of subjectivity: Judgment / Emotion / Argument / Intention / ..."""
labels = ['Agreement', 'Argument', 'Emotion', 'Intention', 'Judgment', 'Others', 'Speculation']
_feature = 'subjectivity-type'
[docs]
class PolarityBlendLexicon(SentimentLexicon):
"""High-accuracy POS/NEG polarity lexicon: the frozen KOSAC seeds blended with
an NSMC-learned lexicon (see ``benchmarks/`` and ``build_shipped_blend.py``).
Unlike the frozen :class:`PolarityLexicon` (5 labels, near-chance out of domain),
this 2-label lexicon is tuned for classification and ships gzipped. Best used
with the analyzer's default multi-scale scorer. Derived data — CC BY-SA (mixes
CC BY-SA KOSAC with CC0 NSMC); see ``data/polarity-blend.NOTICE``.
"""
labels = ['NEG', 'POS']
_feature = 'polarity-blend'
[docs]
class GenericLexicon(SentimentLexicon):
"""A lexicon with user-defined labels (see :meth:`set_labels`)."""
[docs]
def set_labels(self, labels: list):
"""Set the label set for this lexicon."""
self.labels = labels